This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new b9b19b46c7 [docs] Reorganize append table guides and replace diagrams
with SVGs (#9719)
b9b19b46c7 is described below
commit b9b19b46c762467705305bde4b65e154e1f671f3
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Sep 10 19:30:55 2026 +0800
[docs] Reorganize append table guides and replace diagrams with SVGs (#9719)
---
docs/docs/append-table/bucketed.mdx | 170 ++++-----
docs/docs/append-table/incremental-clustering.mdx | 433 +++++++++++-----------
docs/docs/append-table/index.mdx | 186 ++++------
docs/docs/append-table/query-performance.md | 145 ++++++++
docs/docs/append-table/row-level-operations.md | 112 ++++++
docs/docs/append-table/row-tracking.md | 169 ++++++---
docs/docs/append-table/streaming.mdx | 191 ++++++++++
docs/docs/flink/sql-query.mdx | 5 +-
docs/sidebars.js | 5 +-
docs/static/img/append-bucket-order.svg | 59 +++
docs/static/img/append-file-pruning.svg | 63 ++++
docs/static/img/append-incremental-clustering.svg | 60 +++
docs/static/img/append-row-level-storage.svg | 73 ++++
docs/static/img/append-row-tracking.svg | 51 +++
docs/static/img/append-streaming-compaction.svg | 48 +++
docs/static/img/for-queue.png | Bin 339649 -> 0 bytes
docs/static/img/unaware-bucket-topo.png | Bin 717803 -> 0 bytes
17 files changed, 1265 insertions(+), 505 deletions(-)
diff --git a/docs/docs/append-table/bucketed.mdx
b/docs/docs/append-table/bucketed.mdx
index 66b78ad57c..cf33786306 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
-<Tabs groupId="create-bucketed-append">
+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`.
+<Tabs groupId="engine">
<TabItem value="flink" label="Flink">
```sql
-CREATE TABLE my_table (
+CREATE TABLE bucketed_table (
product_id BIGINT,
price DOUBLE,
sales BIGINT
@@ -47,136 +51,86 @@ CREATE TABLE my_table (
```
</TabItem>
+<TabItem value="spark" label="Spark">
-</Tabs>
+```sql
+CREATE TABLE bucketed_table (
+ product_id BIGINT,
+ price DOUBLE,
+ sales BIGINT
+) USING paimon
+TBLPROPERTIES (
+ 'bucket' = '8',
+ 'bucket-key' = 'product_id'
+);
+```
-## Data Skipping
+</TabItem>
+</Tabs>
-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.
-
-
+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**
+
-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:
-
-<table className="configuration table table-bordered">
- <thead>
- <tr>
- <th className="text-left" style={{width: "20%"}}>Key</th>
- <th className="text-left" style={{width: "15%"}}>Default</th>
- <th className="text-left" style={{width: "10%"}}>Type</th>
- <th className="text-left" style={{width: "55%"}}>Description</th>
- </tr>
- </thead>
- <tbody>
- <tr>
- <td><h5>scan.watermark.alignment.group</h5></td>
- <td style={{wordWrap: "break-word"}}>(none)</td>
- <td>String</td>
- <td>A group of sources to align watermarks.</td>
- </tr>
- <tr>
- <td><h5>scan.watermark.alignment.max-drift</h5></td>
- <td style={{wordWrap: "break-word"}}>(none)</td>
- <td>Duration</td>
- <td>Maximal drift to align watermarks, before we pause consuming
from the source/task/partition.</td>
- </tr>
- </tbody>
-</table>
-
-**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 688ee58290..25249a8f16 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:
-<table className="table table-bordered">
- <thead>
- <tr>
- <th className="text-left" style={{width: "20%"}}>Option</th>
- <th className="text-left" style={{width: "10%"}}>Value</th>
- <th className="text-left" style={{width: "5%"}}>Required</th>
- <th className="text-left" style={{width: "10%"}}>Type</th>
- <th className="text-left" style={{width: "55%"}}>Description</th>
- </tr>
- </thead>
- <tbody>
- <tr>
- <td><h5>clustering.incremental</h5></td>
- <td>true</td>
- <td style={{wordWrap: "break-word"}}>Yes</td>
- <td>Boolean</td>
- <td>Must be set to true to enable incremental clustering. Default is
false.</td>
- </tr>
- <tr>
- <td><h5>clustering.columns</h5></td>
- <td>'clustering-columns'</td>
- <td style={{wordWrap: "break-word"}}>Yes</td>
- <td>String</td>
- <td>The clustering columns, in the format 'columnName1,columnName2'. It
is not recommended to use partition keys as clustering keys.</td>
- </tr>
- <tr>
- <td><h5>clustering.strategy</h5></td>
- <td>'zorder' or 'hilbert' or 'order'</td>
- <td style={{wordWrap: "break-word"}}>No</td>
- <td>String</td>
- <td>The 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.</td>
- </tr>
- <tr>
- <td><h5>clustering.incremental.mode</h5></td>
- <td>'global-sort' or 'local-sort'</td>
- <td style={{wordWrap: "break-word"}}>No</td>
- <td>Enum</td>
- <td>The sort mode for incremental clustering compaction. Default is
<code>global-sort</code>. <code>global-sort</code> 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.
<code>local-sort</code> 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 [...]
- </tr>
- </tbody>
-
-</table>
-
-For bucketed append tables (`bucket > 0`), you must also set the following
option:
-
-<table className="table table-bordered">
- <thead>
- <tr>
- <th className="text-left" style={{width: "20%"}}>Option</th>
- <th className="text-left" style={{width: "10%"}}>Value</th>
- <th className="text-left" style={{width: "5%"}}>Required</th>
- <th className="text-left" style={{width: "10%"}}>Type</th>
- <th className="text-left" style={{width: "55%"}}>Description</th>
- </tr>
- </thead>
- <tbody>
- <tr>
- <td><h5>bucket-append-ordered</h5></td>
- <td>false</td>
- <td style={{wordWrap: "break-word"}}>Yes</td>
- <td>Boolean</td>
- <td>Must be set to false for bucketed append tables with incremental
clustering.</td>
- </tr>
- </tbody>
-
-</table>
-
-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:
+
+<Tabs groupId="engine">
+<TabItem value="flink" label="Flink">
+
+```sql
+ALTER TABLE my_table SET (
+ 'clustering.incremental' = 'true',
+ 'clustering.columns' = 'product_id,price'
+);
+```
+
+</TabItem>
+<TabItem value="spark" label="Spark">
```sql
-ALTER TABLE T SET (
+ALTER TABLE my_table SET TBLPROPERTIES (
+ 'clustering.incremental' = 'true',
+ 'clustering.columns' = 'product_id,price'
+);
+```
+
+</TabItem>
+</Tabs>
+
+### 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).
+
+<Tabs groupId="engine">
+<TabItem value="flink" label="Flink">
+
+```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.
+</TabItem>
+<TabItem value="spark" label="Spark">
-**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
+</TabItem>
+</Tabs>
-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:
-<Tabs groupId="incremental-clustering">
+| 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. |
-<TabItem value="spark-sql" label="Spark SQL">
+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.
+
+<Tabs groupId="engine">
+<TabItem value="spark" label="Spark SQL">
```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'
+);
```
-</TabItem>
+Alternatively, `where` accepts a predicate on partition columns. Do not
combine `partitions` and `where` in one call.
-<TabItem value="flink-action" label="Flink Action">
+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'
+);
+```
+
+</TabItem>
+<TabItem value="flink" label="Flink Action">
```bash
<FLINK_HOME>/bin/flink run \
+ -Dexecution.runtime-mode=batch \
/path/to/paimon-flink-action-@@VERSION@@.jar \
compact \
--warehouse <warehouse-path> \
--database <database-name> \
--table <table-name> \
- [--compact_strategy <minor / full>] \
- [--table_conf <table_conf>] \
- [--catalog_conf <paimon-catalog-conf> [--catalog_conf
<paimon-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
-<FLINK_HOME>/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.
+
+</TabItem>
+</Tabs>
+
+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
-<FLINK_HOME>/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.
-</TabItem>
+## How File Selection Works {#implement}
-</Tabs>
+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.
+
+
+
+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:
-<table className="table table-bordered">
- <thead>
- <tr>
- <th className="text-left" style={{width: "20%"}}>Option</th>
- <th className="text-left" style={{width: "10%"}}>Value</th>
- <th className="text-left" style={{width: "5%"}}>Required</th>
- <th className="text-left" style={{width: "10%"}}>Type</th>
- <th className="text-left" style={{width: "55%"}}>Description</th>
- </tr>
- </thead>
- <tbody>
- <tr>
- <td><h5>clustering.history-partition.idle-to-full-sort</h5></td>
- <td>3d</td>
- <td style={{wordWrap: "break-word"}}>Yes</td>
- <td>Duration</td>
- <td>The duration after which a partition without new updates is
considered a historical partition. Default is null.</td>
- </tr>
- <tr>
- <td><h5>clustering.history-partition.limit</h5></td>
- <td>5</td>
- <td style={{wordWrap: "break-word"}}>Yes</td>
- <td>Integer</td>
- <td>The limit of history partition number for automatically performing
full clustering. Default value is 5.</td>
- </tr>
- </tbody>
-
-</table>
-
-
-## 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 152cb9c648..9671911e8f 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.
-<Tabs groupId="create-append-table">
+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.
+
+<Tabs groupId="engine">
<TabItem value="flink" label="Flink">
```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'
);
-```
-
-</TabItem>
-
-</Tabs>
-
-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:
-
-
-
-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:
+</TabItem>
+<TabItem value="spark" label="Spark">
```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.
+</TabItem>
+</Tabs>
-## 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.<filter-type>.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 0000000000..86e9ebe1f7
--- /dev/null
+++ b/docs/docs/append-table/query-performance.md
@@ -0,0 +1,145 @@
+---
+title: "Query Performance"
+sidebar_position: 2
+---
+
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# 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.
+
+
+
+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 0000000000..26945b0ff9
--- /dev/null
+++ b/docs/docs/append-table/row-level-operations.md
@@ -0,0 +1,112 @@
+---
+title: "Row-Level Operations"
+sidebar_position: 5
+---
+
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# 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.
+
+
+
+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 a6737becb1..d3661ec78f 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
---
<!--
@@ -22,44 +22,63 @@ specific language governing permissions and limitations
under the License.
-->
-# 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
+
+
+
+### 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 0000000000..e740f05b05
--- /dev/null
+++ b/docs/docs/append-table/streaming.mdx
@@ -0,0 +1,191 @@
+---
+title: "Streaming"
+sidebar_position: 1
+---
+
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# 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.
+
+
+
+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 ef2e3e9ae6..ec0ddf9f3d 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 086f4dc53f..17e29c1557 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 0000000000..8f95d7e937
--- /dev/null
+++ b/docs/static/img/append-bucket-order.svg
@@ -0,0 +1,59 @@
+<svg xmlns="http://www.w3.org/2000/svg"; width="900" height="416" viewBox="0 0
900 416" role="img" aria-labelledby="title desc">
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<title id="title">Bucket-scoped streaming order</title>
+<desc id="desc">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.</desc>
+<defs><marker id="arrow" viewBox="0 0 10 10" refX="9" refY="5" markerWidth="7"
markerHeight="7" orient="auto"><path d="M 0 0 L 10 5 L 0 10 z"
fill="#526277"/></marker></defs>
+<g font-family="Arial, Helvetica, sans-serif">
+<rect x="1" y="1" width="898" height="414" rx="12" fill="#ffffff"
stroke="#d7dfeb" stroke-width="1.5"/>
+<text x="28" y="42" font-size="27" fill="#172b4d" font-weight="700"
text-anchor="start">Append order is local to a bucket</text>
+<text x="28" y="74" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">One partition · bucket-append-ordered = true · earlier
appends read first</text>
+<rect x="28" y="110" width="650" height="112" rx="8" fill="#f5f7fb"
stroke="#d7dfeb" stroke-width="1.5"/>
+<text x="47" y="143" font-size="20" fill="#172b4d" font-weight="700"
text-anchor="start">Bucket 0</text>
+<rect x="162" y="128" width="136" height="76" rx="8" fill="#eaf2ff"
stroke="#2463b4" stroke-width="1.5"/>
+<text x="230" y="155" font-size="17" fill="#2463b4" font-weight="700"
text-anchor="middle">Commit 1</text>
+<text x="230" y="185" font-size="20" fill="#172b4d" font-weight="400"
text-anchor="middle">A, B</text>
+<path d="M 298 166 H 323" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<rect x="331" y="128" width="136" height="76" rx="8" fill="#eaf2ff"
stroke="#2463b4" stroke-width="1.5"/>
+<text x="399" y="155" font-size="17" fill="#2463b4" font-weight="700"
text-anchor="middle">Commit 2</text>
+<text x="399" y="185" font-size="20" fill="#172b4d" font-weight="400"
text-anchor="middle">C, D</text>
+<path d="M 467 166 H 492" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<rect x="500" y="128" width="136" height="76" rx="8" fill="#eaf2ff"
stroke="#2463b4" stroke-width="1.5"/>
+<text x="568" y="155" font-size="17" fill="#2463b4" font-weight="700"
text-anchor="middle">Commit 3</text>
+<text x="568" y="185" font-size="20" fill="#172b4d" font-weight="400"
text-anchor="middle">E, F</text>
+<path d="M 669 166 H 718" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<rect x="726" y="122" width="146" height="88" rx="8" fill="#e8f7f2"
stroke="#087f6e" stroke-width="1.5"/><text x="799.0" y="156" font-size="19"
fill="#087f6e" font-weight="700" text-anchor="middle">Reader 1</text><text
x="799.0" y="186" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">independent</text>
+<rect x="28" y="241" width="650" height="112" rx="8" fill="#f5f7fb"
stroke="#d7dfeb" stroke-width="1.5"/>
+<text x="47" y="274" font-size="20" fill="#172b4d" font-weight="700"
text-anchor="start">Bucket 1</text>
+<rect x="162" y="259" width="136" height="76" rx="8" fill="#eaf2ff"
stroke="#2463b4" stroke-width="1.5"/>
+<text x="230" y="286" font-size="17" fill="#2463b4" font-weight="700"
text-anchor="middle">Commit 1</text>
+<text x="230" y="316" font-size="20" fill="#172b4d" font-weight="400"
text-anchor="middle">G, H</text>
+<path d="M 298 297 H 323" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<rect x="331" y="259" width="136" height="76" rx="8" fill="#eaf2ff"
stroke="#2463b4" stroke-width="1.5"/>
+<text x="399" y="286" font-size="17" fill="#2463b4" font-weight="700"
text-anchor="middle">Commit 2</text>
+<text x="399" y="316" font-size="20" fill="#172b4d" font-weight="400"
text-anchor="middle">I, J</text>
+<path d="M 467 297 H 492" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<rect x="500" y="259" width="136" height="76" rx="8" fill="#eaf2ff"
stroke="#2463b4" stroke-width="1.5"/>
+<text x="568" y="286" font-size="17" fill="#2463b4" font-weight="700"
text-anchor="middle">Commit 3</text>
+<text x="568" y="316" font-size="20" fill="#172b4d" font-weight="400"
text-anchor="middle">K, L</text>
+<path d="M 669 297 H 718" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<rect x="726" y="253" width="146" height="88" rx="8" fill="#e8f7f2"
stroke="#087f6e" stroke-width="1.5"/><text x="799.0" y="287" font-size="19"
fill="#087f6e" font-weight="700" text-anchor="middle">Reader 2</text><text
x="799.0" y="317" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">independent</text>
+<text x="28" y="393" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">No cross-bucket order: records from the two readers can
interleave downstream.</text>
+</g>
+</svg>
diff --git a/docs/static/img/append-file-pruning.svg
b/docs/static/img/append-file-pruning.svg
new file mode 100644
index 0000000000..07f7fe96dc
--- /dev/null
+++ b/docs/static/img/append-file-pruning.svg
@@ -0,0 +1,63 @@
+<svg xmlns="http://www.w3.org/2000/svg"; width="900" height="583" viewBox="0 0
900 583" role="img" aria-labelledby="title desc">
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<title id="title">File statistics prune more files after clustering</title>
+<desc id="desc">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.</desc>
+<defs><marker id="arrow" viewBox="0 0 10 10" refX="9" refY="5" markerWidth="7"
markerHeight="7" orient="auto"><path d="M 0 0 L 10 5 L 0 10 z"
fill="#526277"/></marker></defs>
+<g font-family="Arial, Helvetica, sans-serif">
+<rect x="1" y="1" width="898" height="581" rx="12" fill="#ffffff"
stroke="#d7dfeb" stroke-width="1.5"/>
+<text x="28" y="42" font-size="27" fill="#172b4d" font-weight="700"
text-anchor="start">Clustering makes file ranges more selective</text>
+<text x="28" y="74" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">Same nine product IDs · predicate: product_id BETWEEN 100
AND 200</text>
+<text x="28" y="117" font-size="19" fill="#93600b" font-weight="700"
text-anchor="start">BEFORE · read all three candidate files</text>
+<rect x="28" y="136" width="270" height="136" rx="8" fill="#fff8ea"
stroke="#ddbf7c" stroke-width="1.5"/>
+<text x="46" y="165" font-size="20" fill="#93600b" font-weight="700"
text-anchor="start">File A</text>
+<text x="46" y="197" font-size="20" fill="#172b4d" font-weight="400"
text-anchor="start">10, 120, 310</text>
+<text x="46" y="231" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">Min/max: 10–310</text>
+<text x="46" y="255" font-size="17" fill="#93600b" font-weight="700"
text-anchor="start">READ · range overlaps</text>
+<rect x="315" y="136" width="270" height="136" rx="8" fill="#fff8ea"
stroke="#ddbf7c" stroke-width="1.5"/>
+<text x="333" y="165" font-size="20" fill="#93600b" font-weight="700"
text-anchor="start">File B</text>
+<text x="333" y="197" font-size="20" fill="#172b4d" font-weight="400"
text-anchor="start">50, 150, 340</text>
+<text x="333" y="231" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">Min/max: 50–340</text>
+<text x="333" y="255" font-size="17" fill="#93600b" font-weight="700"
text-anchor="start">READ · range overlaps</text>
+<rect x="602" y="136" width="270" height="136" rx="8" fill="#fff8ea"
stroke="#ddbf7c" stroke-width="1.5"/>
+<text x="620" y="165" font-size="20" fill="#93600b" font-weight="700"
text-anchor="start">File C</text>
+<text x="620" y="197" font-size="20" fill="#172b4d" font-weight="400"
text-anchor="start">80, 190, 390</text>
+<text x="620" y="231" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">Min/max: 80–390</text>
+<text x="620" y="255" font-size="17" fill="#93600b" font-weight="700"
text-anchor="start">READ · range overlaps</text>
+<path d="M 450 283 V 328" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<text x="468" y="310" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">cluster by product_id</text>
+<text x="28" y="363" font-size="19" fill="#087f6e" font-weight="700"
text-anchor="start">AFTER · read one candidate file</text>
+<rect x="28" y="382" width="270" height="136" rx="8" fill="#f5f7fb"
stroke="#d7dfeb" stroke-width="1.5"/>
+<text x="46" y="411" font-size="20" fill="#526277" font-weight="700"
text-anchor="start">File D</text>
+<text x="46" y="443" font-size="20" fill="#172b4d" font-weight="400"
text-anchor="start">10, 50, 80</text>
+<text x="46" y="477" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">Min/max: 10–80</text>
+<text x="46" y="501" font-size="17" fill="#526277" font-weight="700"
text-anchor="start">SKIP · range outside filter</text>
+<rect x="315" y="382" width="270" height="136" rx="8" fill="#e8f7f2"
stroke="#087f6e" stroke-width="1.5"/>
+<text x="333" y="411" font-size="20" fill="#087f6e" font-weight="700"
text-anchor="start">File E</text>
+<text x="333" y="443" font-size="20" fill="#172b4d" font-weight="400"
text-anchor="start">120, 150, 190</text>
+<text x="333" y="477" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">Min/max: 120–190</text>
+<text x="333" y="501" font-size="17" fill="#087f6e" font-weight="700"
text-anchor="start">READ · range overlaps</text>
+<rect x="602" y="382" width="270" height="136" rx="8" fill="#f5f7fb"
stroke="#d7dfeb" stroke-width="1.5"/>
+<text x="620" y="411" font-size="20" fill="#526277" font-weight="700"
text-anchor="start">File F</text>
+<text x="620" y="443" font-size="20" fill="#172b4d" font-weight="400"
text-anchor="start">310, 340, 390</text>
+<text x="620" y="477" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">Min/max: 310–390</text>
+<text x="620" y="501" font-size="17" fill="#526277" font-weight="700"
text-anchor="start">SKIP · range outside filter</text>
+<text x="28" y="557" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">The reader still evaluates the predicate on rows in the
selected files.</text>
+</g>
+</svg>
diff --git a/docs/static/img/append-incremental-clustering.svg
b/docs/static/img/append-incremental-clustering.svg
new file mode 100644
index 0000000000..dbf9414b90
--- /dev/null
+++ b/docs/static/img/append-incremental-clustering.svg
@@ -0,0 +1,60 @@
+<svg xmlns="http://www.w3.org/2000/svg"; width="900" height="490" viewBox="0 0
900 490" role="img" aria-labelledby="title desc">
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<title id="title">Incremental clustering selects a subset of files</title>
+<desc id="desc">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.</desc>
+<defs><marker id="arrow" viewBox="0 0 10 10" refX="9" refY="5" markerWidth="7"
markerHeight="7" orient="auto"><path d="M 0 0 L 10 5 L 0 10 z"
fill="#526277"/></marker></defs>
+<g font-family="Arial, Helvetica, sans-serif">
+<rect x="1" y="1" width="898" height="488" rx="12" fill="#ffffff"
stroke="#d7dfeb" stroke-width="1.5"/>
+<text x="28" y="42" font-size="27" fill="#172b4d" font-weight="700"
text-anchor="start">Cluster the selected files; keep the rest</text>
+<text x="28" y="74" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">Illustrative incremental run within one partition</text>
+<text x="28" y="113" font-size="17" fill="#526277" font-weight="700"
text-anchor="start">BEFORE</text>
+<text x="632" y="113" font-size="17" fill="#526277" font-weight="700"
text-anchor="start">AFTER</text>
+<rect x="28" y="130" width="255" height="192" rx="8" fill="#fff8ea"
stroke="#ddbf7c" stroke-width="1.5"/>
+<text x="46" y="159" font-size="20" fill="#93600b" font-weight="700"
text-anchor="start">Selected for this run</text>
+<text x="46" y="190" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">Level 0 · new files</text>
+<rect x="46" y="204" width="64" height="42" rx="8" fill="#ffffff"
stroke="#ddbf7c" stroke-width="1.5"/>
+<text x="78" y="232" font-size="20" fill="#93600b" font-weight="700"
text-anchor="middle">A</text>
+<rect x="121" y="204" width="64" height="42" rx="8" fill="#ffffff"
stroke="#ddbf7c" stroke-width="1.5"/>
+<text x="153" y="232" font-size="20" fill="#93600b" font-weight="700"
text-anchor="middle">B</text>
+<rect x="196" y="204" width="64" height="42" rx="8" fill="#ffffff"
stroke="#ddbf7c" stroke-width="1.5"/>
+<text x="228" y="232" font-size="20" fill="#93600b" font-weight="700"
text-anchor="middle">C</text>
+<rect x="46" y="264" width="219" height="40" rx="8" fill="#ffffff"
stroke="#ddbf7c" stroke-width="1.5"/>
+<text x="155" y="291" font-size="18" fill="#93600b" font-weight="400"
text-anchor="middle">Level 1 · existing run</text>
+<rect x="353" y="178" width="200" height="106" rx="8" fill="#eaf2ff"
stroke="#2463b4" stroke-width="1.5"/><text x="453.0" y="212" font-size="19"
fill="#2463b4" font-weight="700" text-anchor="middle">Sort + merge</text><text
x="453.0" y="242" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">clustering columns</text>
+<path d="M 283 230 H 345" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<path d="M 553 230 H 624" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<rect x="632" y="157" width="240" height="152" rx="8" fill="#e8f7f2"
stroke="#087f6e" stroke-width="1.5"/>
+<text x="752" y="188" font-size="20" fill="#087f6e" font-weight="700"
text-anchor="middle">New clustered run</text>
+<rect x="650" y="207" width="92" height="48" rx="8" fill="#ffffff"
stroke="#087f6e" stroke-width="1.5"/>
+<text x="696" y="238" font-size="18" fill="#087f6e" font-weight="400"
text-anchor="middle">File D</text>
+<rect x="758" y="207" width="96" height="48" rx="8" fill="#ffffff"
stroke="#087f6e" stroke-width="1.5"/>
+<text x="806" y="238" font-size="18" fill="#087f6e" font-weight="400"
text-anchor="middle">File E</text>
+<text x="752" y="286" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Target-sized output files</text>
+<rect x="28" y="355" width="255" height="69" rx="8" fill="#f5f7fb"
stroke="#d7dfeb" stroke-width="1.5"/>
+<text x="155" y="385" font-size="19" fill="#172b4d" font-weight="700"
text-anchor="middle">Higher-level run</text>
+<text x="155" y="409" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Not selected</text>
+<path d="M 283 390 H 624" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<text x="452" y="375" font-size="18" fill="#526277" font-weight="400"
text-anchor="middle">No rewrite</text>
+<rect x="632" y="355" width="240" height="69" rx="8" fill="#f5f7fb"
stroke="#d7dfeb" stroke-width="1.5"/>
+<text x="752" y="385" font-size="19" fill="#172b4d" font-weight="700"
text-anchor="middle">Same higher-level run</text>
+<text x="752" y="409" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Reused in the new snapshot</text>
+<text x="28" y="466" font-size="17" fill="#526277" font-weight="400"
text-anchor="start">A full job can also skip already-clustered units. Output
levels depend on the planner.</text>
+</g>
+</svg>
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 0000000000..fe431fa0d5
--- /dev/null
+++ b/docs/static/img/append-row-level-storage.svg
@@ -0,0 +1,73 @@
+<svg xmlns="http://www.w3.org/2000/svg"; width="900" height="633" viewBox="0 0
900 633" role="img" aria-labelledby="title desc">
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<title id="title">Copy on write and deletion vectors produce the same visible
delete result</title>
+<desc id="desc">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.</desc>
+<defs><marker id="arrow" viewBox="0 0 10 10" refX="9" refY="5" markerWidth="7"
markerHeight="7" orient="auto"><path d="M 0 0 L 10 5 L 0 10 z"
fill="#526277"/></marker></defs>
+<g font-family="Arial, Helvetica, sans-serif">
+<rect x="1" y="1" width="898" height="631" rx="12" fill="#ffffff"
stroke="#d7dfeb" stroke-width="1.5"/>
+<text x="28" y="42" font-size="27" fill="#172b4d" font-weight="700"
text-anchor="start">Deleting one row: two physical layouts</text>
+<text x="28" y="74" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">Delete value B · regular append table · current snapshot
shown below</text>
+<rect x="28" y="96" width="844" height="99" rx="8" fill="#f5f7fb"
stroke="#d7dfeb" stroke-width="1.5"/>
+<text x="48" y="132" font-size="20" fill="#172b4d" font-weight="700"
text-anchor="start">Original file F1</text>
+<text x="48" y="167" font-size="17" fill="#526277" font-weight="400"
text-anchor="start">File positions:</text>
+<rect x="289" y="110" width="110" height="42" rx="8" fill="#eaf2ff"
stroke="#2463b4" stroke-width="1.5"/>
+<text x="344" y="139" font-size="20" fill="#2463b4" font-weight="700"
text-anchor="middle">A</text>
+<text x="344" y="179" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">0</text>
+<rect x="429" y="110" width="110" height="42" rx="8" fill="#eaf2ff"
stroke="#2463b4" stroke-width="1.5"/>
+<text x="484" y="139" font-size="20" fill="#2463b4" font-weight="700"
text-anchor="middle">B</text>
+<text x="484" y="179" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">1</text>
+<rect x="569" y="110" width="110" height="42" rx="8" fill="#eaf2ff"
stroke="#2463b4" stroke-width="1.5"/>
+<text x="624" y="139" font-size="20" fill="#2463b4" font-weight="700"
text-anchor="middle">C</text>
+<text x="624" y="179" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">2</text>
+<rect x="709" y="110" width="110" height="42" rx="8" fill="#eaf2ff"
stroke="#2463b4" stroke-width="1.5"/>
+<text x="764" y="139" font-size="20" fill="#2463b4" font-weight="700"
text-anchor="middle">D</text>
+<text x="764" y="179" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">3</text>
+<path d="M 242 195 V 222" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<path d="M 666 195 V 222" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<rect x="28" y="231" width="410" height="256" rx="8" fill="#f8faff"
stroke="#d7dfeb" stroke-width="1.5"/>
+<rect x="462" y="231" width="410" height="256" rx="8" fill="#f0faf7"
stroke="#b8dcd2" stroke-width="1.5"/>
+<text x="48" y="265" font-size="23" fill="#2463b4" font-weight="700"
text-anchor="start">Copy on write</text>
+<text x="482" y="265" font-size="23" fill="#087f6e" font-weight="700"
text-anchor="start">Deletion vectors</text>
+<text x="48" y="298" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">Write replacement file F2</text>
+<text x="482" y="298" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">Keep data file F1 unchanged</text>
+<rect x="48" y="316" width="108" height="50" rx="8" fill="#eaf2ff"
stroke="#2463b4" stroke-width="1.5"/>
+<text x="102" y="348" font-size="21" fill="#2463b4" font-weight="700"
text-anchor="middle">A</text>
+<rect x="171" y="316" width="108" height="50" rx="8" fill="#eaf2ff"
stroke="#2463b4" stroke-width="1.5"/>
+<text x="225" y="348" font-size="21" fill="#2463b4" font-weight="700"
text-anchor="middle">C</text>
+<rect x="294" y="316" width="108" height="50" rx="8" fill="#eaf2ff"
stroke="#2463b4" stroke-width="1.5"/>
+<text x="348" y="348" font-size="21" fill="#2463b4" font-weight="700"
text-anchor="middle">D</text>
+<rect x="482" y="316" width="82" height="50" rx="8" fill="#eaf2ff"
stroke="#2463b4" stroke-width="1.5"/>
+<text x="523" y="348" font-size="21" fill="#2463b4" font-weight="700"
text-anchor="middle">A</text>
+<rect x="577" y="316" width="82" height="50" rx="8" fill="#eaf2ff"
stroke="#2463b4" stroke-width="1.5"/>
+<text x="618" y="348" font-size="21" fill="#2463b4" font-weight="700"
text-anchor="middle">B</text>
+<rect x="672" y="316" width="82" height="50" rx="8" fill="#eaf2ff"
stroke="#2463b4" stroke-width="1.5"/>
+<text x="713" y="348" font-size="21" fill="#2463b4" font-weight="700"
text-anchor="middle">C</text>
+<rect x="767" y="316" width="82" height="50" rx="8" fill="#eaf2ff"
stroke="#2463b4" stroke-width="1.5"/>
+<text x="808" y="348" font-size="21" fill="#2463b4" font-weight="700"
text-anchor="middle">D</text>
+<text x="48" y="409" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">The current snapshot references F2</text>
+<text x="48" y="438" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">in place of F1.</text>
+<rect x="482" y="387" width="370" height="59" rx="8" fill="#fff0f1"
stroke="#c97581" stroke-width="1.5"/>
+<text x="667" y="423" font-size="19" fill="#a32b3a" font-weight="700"
text-anchor="middle">Deletion vector: F1, position 1</text>
+<text x="482" y="469" font-size="17" fill="#526277" font-weight="400"
text-anchor="start">The reader excludes the marked position.</text>
+<rect x="28" y="510" width="844" height="65" rx="8" fill="#e8f7f2"
stroke="#087f6e" stroke-width="1.5"/>
+<text x="450" y="551" font-size="23" fill="#087f6e" font-weight="700"
text-anchor="middle">Visible result in both cases: A, C, D</text>
+<text x="28" y="608" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">Replaced files can remain referenced by older snapshots
until expiration.</text>
+</g>
+</svg>
diff --git a/docs/static/img/append-row-tracking.svg
b/docs/static/img/append-row-tracking.svg
new file mode 100644
index 0000000000..6a2a377aba
--- /dev/null
+++ b/docs/static/img/append-row-tracking.svg
@@ -0,0 +1,51 @@
+<svg xmlns="http://www.w3.org/2000/svg"; width="900" height="461" viewBox="0 0
900 461" role="img" aria-labelledby="title desc">
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<title id="title">Row ID stays stable while row version changes on
update</title>
+<desc id="desc">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.</desc>
+<defs><marker id="arrow" viewBox="0 0 10 10" refX="9" refY="5" markerWidth="7"
markerHeight="7" orient="auto"><path d="M 0 0 L 10 5 L 0 10 z"
fill="#526277"/></marker></defs>
+<g font-family="Arial, Helvetica, sans-serif">
+<rect x="1" y="1" width="898" height="459" rx="12" fill="#ffffff"
stroke="#d7dfeb" stroke-width="1.5"/>
+<text x="28" y="42" font-size="27" fill="#172b4d" font-weight="700"
text-anchor="start">Row identity and row version</text>
+<text x="28" y="74" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">One row in a regular row-tracking table · illustrative
snapshot IDs</text>
+<rect x="28" y="111" width="236" height="213" rx="8" fill="#eaf2ff"
stroke="#2463b4" stroke-width="1.5"/>
+<text x="46" y="145" font-size="20" fill="#2463b4" font-weight="700"
text-anchor="start">INSERT</text>
+<text x="46" y="174" font-size="17" fill="#526277" font-weight="400"
text-anchor="start">Snapshot 1</text>
+<text x="46" y="210" font-size="20" fill="#172b4d" font-weight="700"
text-anchor="start">data = a</text>
+<text x="46" y="247" font-size="19" fill="#172b4d" font-weight="400"
text-anchor="start">_ROW_ID = 0</text>
+<text x="46" y="287" font-size="16" fill="#172b4d" font-weight="400"
text-anchor="start">_SEQUENCE_NUMBER = 1</text>
+<rect x="332" y="111" width="236" height="213" rx="8" fill="#fff8ea"
stroke="#93600b" stroke-width="1.5"/>
+<text x="350" y="145" font-size="20" fill="#93600b" font-weight="700"
text-anchor="start">UPDATE</text>
+<text x="350" y="174" font-size="17" fill="#526277" font-weight="400"
text-anchor="start">Snapshot 2</text>
+<text x="350" y="210" font-size="20" fill="#172b4d" font-weight="700"
text-anchor="start">data = a2</text>
+<text x="350" y="247" font-size="19" fill="#172b4d" font-weight="400"
text-anchor="start">_ROW_ID = 0</text>
+<text x="350" y="287" font-size="16" fill="#172b4d" font-weight="400"
text-anchor="start">_SEQUENCE_NUMBER = 2</text>
+<rect x="636" y="111" width="236" height="213" rx="8" fill="#e8f7f2"
stroke="#087f6e" stroke-width="1.5"/>
+<text x="654" y="145" font-size="20" fill="#087f6e" font-weight="700"
text-anchor="start">COMPACT</text>
+<text x="654" y="174" font-size="17" fill="#526277" font-weight="400"
text-anchor="start">Snapshot 3</text>
+<text x="654" y="210" font-size="20" fill="#172b4d" font-weight="700"
text-anchor="start">data = a2</text>
+<text x="654" y="247" font-size="19" fill="#172b4d" font-weight="400"
text-anchor="start">_ROW_ID = 0</text>
+<text x="654" y="287" font-size="16" fill="#172b4d" font-weight="400"
text-anchor="start">_SEQUENCE_NUMBER = 2</text>
+<path d="M 264 219 H 324" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<path d="M 568 219 H 628" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<text x="28" y="367" font-size="20" fill="#2463b4" font-weight="700"
text-anchor="start">Same ID across all three snapshots.</text>
+<text x="28" y="401" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">The update changes the row version; ordinary compaction
preserves it.</text>
+<text x="28" y="434" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">A later DELETE removes the row from the current
result.</text>
+</g>
+</svg>
diff --git a/docs/static/img/append-streaming-compaction.svg
b/docs/static/img/append-streaming-compaction.svg
new file mode 100644
index 0000000000..cb234e247f
--- /dev/null
+++ b/docs/static/img/append-streaming-compaction.svg
@@ -0,0 +1,48 @@
+<svg xmlns="http://www.w3.org/2000/svg"; width="900" height="460" viewBox="0 0
900 460" role="img" aria-labelledby="title desc">
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<title id="title">Flink streaming compaction</title>
+<desc id="desc">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.</desc>
+<defs><marker id="arrow" viewBox="0 0 10 10" refX="9" refY="5" markerWidth="7"
markerHeight="7" orient="auto"><path d="M 0 0 L 10 5 L 0 10 z"
fill="#526277"/></marker></defs>
+<g font-family="Arial, Helvetica, sans-serif">
+<rect x="1" y="1" width="898" height="458" rx="12" fill="#ffffff"
stroke="#d7dfeb" stroke-width="1.5"/>
+<text x="28" y="42" font-size="27" fill="#172b4d" font-weight="700"
text-anchor="start">Flink streaming compaction</text>
+<text x="28" y="73" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">Unaware-bucket append · background compaction enabled</text>
+<rect x="345" y="101" width="362" height="169" rx="8" fill="#f0faf7"
stroke="#b8dcd2" stroke-width="1.5"/>
+<text x="526" y="126" font-size="18" fill="#087f6e" font-weight="700"
text-anchor="middle">Background compaction operators</text>
+<rect x="28" y="148" width="128" height="92" rx="8" fill="#eaf2ff"
stroke="#2463b4" stroke-width="1.5"/><text x="92.0" y="182" font-size="19"
fill="#2463b4" font-weight="700" text-anchor="middle">Input</text><text
x="92.0" y="212" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">insert-only</text>
+<rect x="190" y="148" width="126" height="92" rx="8" fill="#eaf2ff"
stroke="#2463b4" stroke-width="1.5"/><text x="253.0" y="182" font-size="19"
fill="#2463b4" font-weight="700" text-anchor="middle">Writer</text><text
x="253.0" y="212" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">new files</text>
+<rect x="363" y="148" width="155" height="92" rx="8" fill="#e8f7f2"
stroke="#087f6e" stroke-width="1.5"/><text x="440.5" y="182" font-size="19"
fill="#087f6e" font-weight="700" text-anchor="middle">Coordinator</text><text
x="440.5" y="212" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">parallelism = 1</text>
+<rect x="552" y="148" width="137" height="92" rx="8" fill="#e8f7f2"
stroke="#087f6e" stroke-width="1.5"/><text x="620.5" y="182" font-size="19"
fill="#087f6e" font-weight="700" text-anchor="middle">Workers</text><text
x="620.5" y="212" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">async rewrite</text>
+<rect x="733" y="148" width="139" height="92" rx="8" fill="#eaf2ff"
stroke="#2463b4" stroke-width="1.5"/><text x="802.5" y="182" font-size="19"
fill="#2463b4" font-weight="700" text-anchor="middle">Committer</text><text
x="802.5" y="212" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">snapshots</text>
+<path d="M 156 194 H 182" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<path d="M 316 194 H 355" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<path d="M 518 194 H 544" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<path d="M 689 194 H 725" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<text x="28" y="272" font-size="17" fill="#526277" font-weight="400"
text-anchor="start">No bucket-key shuffle</text>
+<rect x="363" y="326" width="509" height="73" rx="8" fill="#f5f7fb"
stroke="#d7dfeb" stroke-width="1.5"/>
+<text x="617" y="355" font-size="21" fill="#172b4d" font-weight="700"
text-anchor="middle">Table snapshots and data files</text>
+<text x="617" y="381" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Committed files are candidates for later compaction</text>
+<path d="M 440 326 V 279" fill="none" stroke="#526277" stroke-width="2"
stroke-dasharray="6 5" marker-end="url(#arrow)"/>
+<text x="458" y="302" font-size="17" fill="#526277" font-weight="400"
text-anchor="start">discover files</text>
+<path d="M 802 240 V 318" fill="none" stroke="#526277" stroke-width="2"
stroke-dasharray="6 5" marker-end="url(#arrow)"/>
+<text x="819" y="285" font-size="17" fill="#526277" font-weight="400"
text-anchor="start">commit</text>
+<text x="28" y="436" font-size="17" fill="#526277" font-weight="400"
text-anchor="start">write-only = true removes the background operators;
pre-commit compaction is separate.</text>
+</g>
+</svg>
diff --git a/docs/static/img/for-queue.png b/docs/static/img/for-queue.png
deleted file mode 100644
index 5e453b7c1f..0000000000
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 f530fc4a22..0000000000
Binary files a/docs/static/img/unaware-bucket-topo.png and /dev/null differ
