diff --git a/website/docs/maintenance/operations/upgrade-notes-1.0.md b/website/docs/maintenance/operations/upgrade-notes-1.0.md index 5850b707b8..4bac76be12 100644 --- a/website/docs/maintenance/operations/upgrade-notes-1.0.md +++ b/website/docs/maintenance/operations/upgrade-notes-1.0.md @@ -50,3 +50,43 @@ After `datalake.enabled` is later set to `true`, tables created under this confi If your existing deployment or internal scripts only set `datalake.format`, they will continue to work with the legacy behavior as long as `datalake.enabled` remains unset. For new configuration examples and operational guidance, we recommend explicitly configuring `datalake.enabled` together with `datalake.format`. + +## Lake Table Schema Changes (FIP-27) + +Starting from this version, Fluss creates lake tables with a **clean** physical schema that contains only the user-defined columns. Earlier versions appended three trailing system columns (`__bucket`, `__offset`, `__timestamp`) to every lake table; these are no longer added to newly created tables. + +This applies to the Paimon and Iceberg lake formats. The Hudi lake storage was never exposed in a publicly released version, so it only ever uses the clean layout and the compatibility considerations below do not apply to it. + +### Clean and Legacy Layouts + +- **Clean layout**: newly created lake tables contain only user columns. +- **Legacy layout**: lake tables created by earlier Fluss versions still carry the three trailing system columns. + +Existing legacy tables are **not** migrated and remain fully readable and writable. Fluss detects the layout directly from the physical schema — a table is treated as legacy when it carries the system columns, and clean otherwise — so both layouts are supported side by side. The tiering service keeps writing the legacy layout for a table that already has the system columns, and writes the clean layout for newly created tables. + +The names `__bucket`, `__offset`, and `__timestamp` remain reserved for Fluss internal use. System columns are disabled by default; any future opt-in behavior to re-enable them is outside the scope of FIP-27. + +### Compatibility Matrix + +| Component | Legacy tables | Clean tables | +|-----------|---------------|--------------| +| New lake storage plugins / lake-reading Flink connectors | Readable | Readable | +| New tiering service | Writes legacy layout | Writes clean layout | +| Old tiering service | Supported | **Not supported** — must not process clean tables | +| Old Flink connectors using `FULL` startup mode | Readable | **Not readable** | + +Old Flink connectors that use `FULL` startup mode assume the presence of the system columns and therefore cannot read newly created clean lake tables. Note that `FULL` is the **default** value of `scan.startup.mode`, so an old connector that does not explicitly set a startup mode is also affected — when auditing jobs before an upgrade, do not look only for jobs that explicitly configure `FULL`. A lake-reading Flink connector must be upgraded together with its matching lake storage plugin. + +### Required Upgrade Order + +To move to a version that creates clean lake tables safely, upgrade the components in this order: + +1. **Lake-reading Flink connectors and lake storage plugins** — so that readers can handle both the legacy and clean layouts before any clean table exists. +2. **Tiering service** — so that it starts producing clean tables only after the readers can consume them. +3. **Fluss cluster**. + +Upgrading in a different order can leave an old reader or an old tiering service facing a clean table it cannot handle. + +### Rollback Limitations + +Once a clean lake table has been created, rolling back to an older tiering service or older lake-reading connectors is **not** safe: those components assume the system columns are present and cannot correctly read or write the clean table. Plan the upgrade with this in mind, since a clean table cannot be transparently rolled back to the legacy layout. diff --git a/website/docs/streaming-lakehouse/datalake-formats/hudi.md b/website/docs/streaming-lakehouse/datalake-formats/hudi.md index 7292f82f94..a94ad8a7cd 100644 --- a/website/docs/streaming-lakehouse/datalake-formats/hudi.md +++ b/website/docs/streaming-lakehouse/datalake-formats/hudi.md @@ -137,6 +137,10 @@ The schema of the Hudi table matches the Fluss table schema, containing only the The names `__bucket`, `__offset`, and `__timestamp` are reserved for Fluss internal use, so do not use user columns with these names. Hudi metadata column names starting with `_hoodie_` are also reserved. +:::note +Unlike Paimon and Iceberg, the Hudi lake storage was never exposed in a publicly released Fluss version, so there are no legacy Hudi tables carrying system columns. Hudi therefore only ever uses the clean layout, and the legacy-table rolling-upgrade considerations in the [Upgrade Notes](../../maintenance/operations/upgrade-notes-1.0.md) do not apply to Hudi. +::: + ### Primary Key Tables Primary-key Fluss tables are mapped to Hudi Merge-On-Read tables: diff --git a/website/docs/streaming-lakehouse/datalake-formats/iceberg.md b/website/docs/streaming-lakehouse/datalake-formats/iceberg.md index 55a50e058e..205e790123 100644 --- a/website/docs/streaming-lakehouse/datalake-formats/iceberg.md +++ b/website/docs/streaming-lakehouse/datalake-formats/iceberg.md @@ -218,8 +218,17 @@ Use Iceberg-specific configurations as parameters when starting the Flink tierin When a Fluss table is created or altered with the option `'table.datalake.enabled' = 'true'` and configured with Iceberg as the datalake format, Fluss will automatically create a corresponding Iceberg table with the same table path. -The schema of the Iceberg table matches that of the Fluss table, except for the addition of three system columns at the end: `__bucket`, `__offset`, and `__timestamp`. -These system columns help Fluss clients consume data from Iceberg in a streaming fashion, such as seeking by a specific bucket using an offset or timestamp. +Newly created Iceberg tables (**clean** tables) contain only the user-defined columns of the Fluss table. Fluss no longer appends the `__bucket`, `__offset`, and `__timestamp` system columns to the physical schema, and a clean table therefore has no `__bucket` partitioning or `__offset` sort order. + +:::note +The **Primary Key Tables**, **Log Tables**, and **Partitioned Tables** sections below describe the mapping for the current **clean** layout: only user columns, the bucket transform on the user bucket key (if any), the identity transform on user partition keys, and no sort order. + +Iceberg tables created by earlier Fluss versions (**legacy** tables) additionally carry the three trailing `__bucket`/`__offset`/`__timestamp` columns, an `identity(__bucket)` partition for bucket-unaware tables, and an `ASC(__offset)` sort order. These tables are **not** migrated and remain fully readable and writable. Fluss detects the layout from the physical schema — a table is treated as legacy when it carries the system columns, and clean otherwise — so both layouts are supported side by side without any manual migration. + +The names `__bucket`, `__offset`, and `__timestamp` remain reserved for Fluss internal use, so user columns must not use these names. + +For the rolling-upgrade requirements when moving to a Fluss version that creates clean tables, see [Upgrade Notes](../../maintenance/operations/upgrade-notes-1.0.md). +::: ### Basic Configuration @@ -262,7 +271,6 @@ CREATE TABLE fluss_order_with_lake ( ) WITH ( 'table.datalake.enabled' = 'true', 'table.datalake.freshness' = '30s', - 'table.datalake.auto-compaction' = 'true', 'iceberg.write.format.default' = 'orc', 'iceberg.commit.retry.num-retries' = '5' ); @@ -274,8 +282,7 @@ Primary key tables in Fluss are mapped to Iceberg tables with: - **Primary key constraints**: The Iceberg table maintains the same primary key definition - **Merge-on-read (MOR) strategy**: Updates and deletes are handled efficiently using Iceberg's MOR capabilities -- **Bucket partitioning**: Automatically partitioned by the primary key using Iceberg's bucket transform with the bucket num of Fluss to align with Fluss -- **Sorted by system column `__offset`**: Sorted by the system column `__offset` (which is derived from the Fluss change log) to preserve the data order and facilitate mapping back to the original Fluss change log +- **Bucket partitioning**: Automatically partitioned by the primary key (the bucket key) using Iceberg's bucket transform with the bucket num of Fluss to align with Fluss ```sql title="Primary Key Table Example" CREATE TABLE user_profiles ( @@ -300,12 +307,8 @@ CREATE TABLE user_profiles ( email STRING, last_login TIMESTAMP, profile_data STRING, - __bucket INT, - __offset BIGINT, - __timestamp TIMESTAMP_LTZ, PRIMARY KEY (user_id) NOT ENFORCED -) PARTITIONED BY (bucket(user_id, 4)) -SORTED BY (__offset ASC); +) PARTITIONED BY (bucket(user_id, 4)); ``` ### Log Tables @@ -314,10 +317,7 @@ The table mapping for Fluss log tables varies depending on whether the bucket ke #### No Bucket Key -Log tables without bucket in Fluss are mapped to Iceberg tables with: - -- **Identity partitioning**: Using identity partitioning on the `__bucket` system column, which enables seeking to the data files in Iceberg if a specified Fluss bucket is given -- **Sorted by system column `__offset`**: Sorted by the system column `__offset` (which is derived from the Fluss log data) to preserve the data order and facilitate mapping back to the original Fluss log data +Log tables without a bucket key in Fluss are mapped to unpartitioned Iceberg tables containing only the user columns. ```sql title="Log Table without Bucket Key" CREATE TABLE access_logs ( @@ -337,12 +337,8 @@ CREATE TABLE access_logs ( timestamp TIMESTAMP, user_id BIGINT, action STRING, - ip_address STRING, - __bucket INT, - __offset BIGINT, - __timestamp TIMESTAMP_LTZ -) PARTITIONED BY (IDENTITY(__bucket)) -SORTED BY (__offset ASC); + ip_address STRING +); ``` #### Single Bucket Key @@ -350,7 +346,6 @@ SORTED BY (__offset ASC); Log tables with one bucket key in Fluss are mapped to Iceberg tables with: - **Bucket partitioning**: Automatically partitioned by the bucket key using Iceberg's bucket transform with the bucket num of Fluss to align with Fluss -- **Sorted by system column `__offset`**: Sorted by the system column `__offset` (which is derived from the Fluss log data) to preserve the data order and facilitate mapping back to the original Fluss log data ```sql title="Log Table with Bucket Key" CREATE TABLE order_events ( @@ -371,12 +366,8 @@ CREATE TABLE order_events ( order_id BIGINT, item_id BIGINT, amount INT, - event_time TIMESTAMP, - __bucket INT, - __offset BIGINT, - __timestamp TIMESTAMP_LTZ -) PARTITIONED BY (bucket(order_id, 5)) -SORTED BY (__offset ASC); + event_time TIMESTAMP +) PARTITIONED BY (bucket(order_id, 5)); ``` ### Partitioned Tables @@ -405,17 +396,13 @@ CREATE TABLE daily_sales ( amount DECIMAL(10,2), customer_id BIGINT, sale_date STRING, - __bucket INT, - __offset BIGINT, - __timestamp TIMESTAMP_LTZ, PRIMARY KEY (sale_id) NOT ENFORCED -) PARTITIONED BY (IDENTITY(sale_date), bucket(sale_id, 4)) -SORTED BY (__offset ASC); +) PARTITIONED BY (IDENTITY(sale_date), bucket(sale_id, 4)); ``` ### System Columns -All Iceberg tables created by Fluss include three system columns: +**Legacy** Iceberg tables (created by earlier Fluss versions) include three trailing system columns. Newly created **clean** tables do not include them. | Column | Type | Description | |---------------|---------------|-----------------------------------------------| @@ -564,7 +551,9 @@ Tiering-managed auto compaction is currently only supported for legacy Iceberg l #### Configuration -```sql title="Flink SQL" +As noted above, `table.datalake.auto-compaction` only takes effect for legacy tables. The following shows the option syntax; it is a no-op on a newly created clean table. + +```sql title="Flink SQL (legacy tables only)" CREATE TABLE example_table ( id BIGINT, data STRING, diff --git a/website/docs/streaming-lakehouse/datalake-formats/paimon.md b/website/docs/streaming-lakehouse/datalake-formats/paimon.md index de56df35db..9f1144575b 100644 --- a/website/docs/streaming-lakehouse/datalake-formats/paimon.md +++ b/website/docs/streaming-lakehouse/datalake-formats/paimon.md @@ -43,8 +43,15 @@ Verify downloaded JARs using the [verification instructions](/downloads#verifyin For general guidance on configuring Paimon as the lakehouse storage, you can refer to [Deploying Streaming Lakehouse](../../install-deploy/deploying-streaming-lakehouse.md) documentation. When starting the tiering service, make sure to use Paimon-specific configurations as parameters. When a table is created or altered with the option `'table.datalake.enabled' = 'true'`, Fluss will automatically create a corresponding Paimon table with the same table path. -The schema of the Paimon table matches that of the Fluss table, except for the addition of three system columns at the end: `__bucket`, `__offset`, and `__timestamp`. -These system columns help Fluss clients consume data from Paimon in a streaming fashion, such as seeking by a specific bucket using an offset or timestamp. +Newly created Paimon tables (**clean** tables) contain only the user-defined columns of the Fluss table. Fluss no longer appends the `__bucket`, `__offset`, and `__timestamp` system columns to the physical schema. + +:::note +Paimon tables created by earlier Fluss versions (**legacy** tables) still carry the three trailing system columns. These tables are **not** migrated and remain fully readable and writable. Fluss detects the layout from the physical schema — a table is treated as legacy when it carries the system columns, and clean otherwise — so both layouts are supported side by side without any manual migration. + +The names `__bucket`, `__offset`, and `__timestamp` remain reserved for Fluss internal use, so user columns must not use these names. + +For the rolling-upgrade requirements when moving to a Fluss version that creates clean tables, see [Upgrade Notes](../../maintenance/operations/upgrade-notes-1.0.md). +::: ```sql title="Flink SQL" USE CATALOG fluss_catalog;