yunfengzhou-hub · Oct 23, 2026 · Oct 27, 2026
diff --git a/docs/docs/concepts/lake-stream.md b/docs/docs/concepts/lake-stream.md
@@ -0,0 +1,97 @@
+---
+title: "Lake-Stream Mode"
+sidebar_position: 11
+---
+
+<!--
+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.
+-->
+
+# Lake-Stream Mode
+
+Lake-stream mode keeps a table's history in Paimon and serves its most recent writes from an
+external streaming store, so a query can return data that is seconds old instead of waiting for the
+next snapshot. Enable it on an existing table when snapshot-level freshness is too slow and you
+already run a supported store. The table keeps its name, its schema, and its historical data.
+
+A table is a lake-stream table when its `lakestream.enabled` option is `true`. Paimon then routes
+each read and write by table name, and the store's tiering service moves data into Paimon in the
+background.
+
+Enabling the mode needs engine-specific catalog configuration, so each engine has its own page.
+[Flink](../flink/lake-stream) is supported today, and Spark is not supported yet. Fluss is the first
+supported store, and [Use Fluss as the Store](../flink/lake-stream#use-fluss-as-the-store) walks
+through it end to end.
+
+## Read a Lake-Stream Table
+
+The name you query decides which data you get, and both names resolve to the same table.
+
+| Name | What the read returns |
+| --- | --- |
+| `my_table` | Paimon snapshots only. Existing queries keep their current behavior and visibility. |
+| `my_table$rt` | Paimon snapshots plus the records the store has not tiered yet. |
+
+Query `my_table$rt` when you need the freshest data, and keep `my_table` for queries that must
+reproduce a committed snapshot.
+
+`$rt` is the same table read through the store rather than a second table. Writing to `my_table$rt`
+is rejected; write to `my_table` and let the routing above decide where it lands. Querying
+`my_table$rt` on a table without `lakestream.enabled` fails as an unknown table, and time travel
+does not apply to it. Read a historical snapshot through `my_table` instead.
+
+## Write to a Lake-Stream Table
+
+Regular writes go to the store, and the store's tiering service moves them into Paimon afterwards.
+This covers streaming and batch inserts, and `DELETE` and `UPDATE` where the store supports them.
+
+`INSERT OVERWRITE` is not supported on a lake-stream table: every write to such a table is served by
+the store, so an overwrite never reaches Paimon.
+
+## Change a Lake-Stream Table
+
+Paimon rejects a table rename while `lakestream.enabled` is `true`. The table name is how Paimon
+finds the matching table in the store, so a rename would break that lookup.
+
+Paimon also rejects replacing the table, which is what `CREATE OR REPLACE TABLE` does. Replacing a
+table can truncate data the store has not tiered yet, and it can drop the `lakestream.enabled`
+option without the restart that [disabling the mode](#disable-lake-stream-mode) requires.
+
+Disable lake-stream mode first if you need to rename or replace the table.
+
+Column changes are allowed, and they update the Paimon table only. The store's tiering service
+reconciles its own schema afterwards, so the two schemas can differ briefly. Where you can, make
+column changes through the store's catalog to keep the two schemas aligned.
+
+Dropping a lake-stream table removes it from Paimon and leaves the store's table behind. The store
+owns that table's lifecycle, so remove it through the store if you no longer need it.
+
+## Disable Lake-Stream Mode
+
+Set the table option back to `false`.
+
+```sql
+ALTER TABLE my_table SET ('lakestream.enabled' = 'false');
+```
+
+Reads and writes then use Paimon only, and `my_table$rt` stops resolving. Restart running write
+pipelines without state so that they plan against the Paimon sink again.
+
+Paimon does not wait for the store to finish tiering. Records that are still only in the store when
+you disable the mode do not appear in later Paimon reads, so confirm that tiering has caught up if
+you need complete results.
diff --git a/docs/docs/flink/lake-stream.md b/docs/docs/flink/lake-stream.md
@@ -0,0 +1,145 @@
+---
+title: "Lake-Stream Mode"
+---
+
+<!--
+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.
+-->
+
+# Lake-Stream Mode
+
+The Flink connector routes the reads and writes of a lake-stream table through the streaming store
+you name on the catalog. [Lake-Stream Mode](../concepts/lake-stream) explains what the mode does to
+a table.
+
+## Enable Lake-Stream Mode
+
+A lake-stream table needs three things: a table of the same database and name in the store, the
+`lakestream.enabled` option on the Paimon table, and a Paimon catalog that names the store.
+
+Create the Paimon catalog with `lakestream.store` and the store's connection options.
+
+```sql
+CREATE CATALOG `paimon_catalog` WITH (
+    'type' = 'paimon',
+    'warehouse' = 'hdfs:///path/to/warehouse',
+    'lakestream.store' = 'fluss',
+    'fluss.bootstrap.servers' = 'fluss-server:9123'
+);
+```
+
+`lakestream.store` names the store implementation. Prefix every store connection option with that
+identifier, and Paimon removes the prefix before it passes the options on, so
+`fluss.bootstrap.servers` reaches Fluss as `bootstrap.servers`. See
+[CatalogOptions](../maintenance/configurations#catalogoptions) for the catalog reference.
+
+Paimon does not create the store-side table and does not move a table's existing data into it.
+Create the store table yourself, in a catalog of the store;
+[Use Fluss as the Store](#use-fluss-as-the-store) shows how for Fluss.
+
+Stop the pipelines that write to the table before you change either side, and restart them without
+state once both sides exist, so that Flink plans them against the store sink. A write that lands
+while the store side is missing, or a writer that resumes from state saved against the Paimon sink,
+can leave the store and Paimon holding different data for the same keys.
+
+A catalog created without `lakestream.store` cannot serve a lake-stream table. `my_table$rt` fails
+as an unknown table, and a write to `my_table` is rejected instead of quietly going to Paimon, which
+would leave the store and Paimon holding different data for the same keys. Tables that are not
+lake-stream tables are unaffected. Add `lakestream.store` and the store's connection options to the
+catalog, or [disable the mode](../concepts/lake-stream#disable-lake-stream-mode) on the table first.
+
+## Write to a Lake-Stream Table
+
+Your write statements do not change. Flink reads `lakestream.enabled` from the table when it plans
+the statement and routes the write accordingly.
+
+```sql
+INSERT INTO my_table SELECT * FROM source_table;
+```
+
+A Flink CDC pipeline does not plan its writes through the SQL sink, so it writes straight to Paimon
+and bypasses the store. Confirm that the data it writes has already expired from the store: while
+the store still holds records for the same primary keys, the two can disagree about them, and later
+updates from the store may then use stale state. See
+[Write to a Lake-Stream Table](../concepts/lake-stream#write-to-a-lake-stream-table).
+
+## Read the Real-Time Name
+
+`my_table$rt` is the same table read through the store, not a second table. Write the name in
+backticks:
+
+```sql
+-- Reads only data that Paimon has committed.
+SELECT * FROM my_table;
+
+-- Combines committed Paimon data with the store's un-tiered records.
+SELECT * FROM `my_table$rt`;
+```
+
+Paimon connects to the store on the first lake-stream read or write. An unknown store identifier or
+a connection option the store requires fails at that point with a configuration error.
+
+## Use Fluss as the Store
+
+Fluss is a streaming storage system for the lakehouse, and can serve as a Paimon Lakestream Store.
+Lake-stream mode requires Fluss 1.0 or later.
+
+Set up a Fluss cluster with Paimon as its lakehouse storage first; the Fluss documentation covers
+[deploying the streaming lakehouse](https://fluss.apache.org/docs/install-deploy/deploying-streaming-lakehouse/)
+and [Paimon as a datalake format](https://fluss.apache.org/docs/streaming-lakehouse/datalake-formats/paimon/).
+
+Then create the store-side table and the Paimon table that points at it. The two have to agree on
+the database name and the table name, because that name is how Paimon finds the table in the store.
+
+```sql
+CREATE CATALOG `fluss_store` WITH (
+    'type' = 'fluss',
+    'bootstrap.servers' = 'fluss-server:9123'
+);
+
+CREATE TABLE fluss_store.my_db.my_table (id INT);
+
+CREATE TABLE paimon_catalog.my_db.my_table (id INT) WITH (
+    'lakestream.enabled' = 'true'
+);
+```
+
+Writes and reads are then the ordinary ones, in the Paimon catalog:
+
+```sql
+INSERT INTO my_table SELECT * FROM source_table;
+
+SELECT * FROM `my_table$rt`;
+```
+
+:::note
+A Fluss log table is unbounded, so Flink rejects a batch query on `my_table$rt` as a query on an
+unbounded table. Bound the read with scan hints. The example below reads the log from the beginning
+up to the offset it had when the query started.
+:::
+
+```sql
+SELECT * FROM `my_table$rt`
+/*+ OPTIONS('scan.startup.mode' = 'earliest', 'scan.bounded.mode' = 'latest-offset') */;
+```
+
+Records reach Paimon only through the Fluss tiering service, which is a separate component from the
+Fluss cluster that serves `my_table$rt`. Until the tiering service runs, the records stay in Fluss:
+`my_table$rt` returns them and `my_table` returns nothing. The Fluss documentation covers the
+[tiering service](https://fluss.apache.org/docs/streaming-lakehouse/tiering-service/) and
+[union reads](https://fluss.apache.org/docs/streaming-lakehouse/union-read/).
diff --git a/docs/generated/catalog_configuration.html b/docs/generated/catalog_configuration.html
@@ -122,6 +122,12 @@
             <td>Boolean</td>
             <td>Whether to support format tables, format table corresponds to a regular csv, parquet or orc table, allowing read and write operations. However, during these processes, it does not connect to the metastore; hence, newly added partitions will not be reflected in the metastore and need to be manually added as separate partition operations.</td>
         </tr>
+        <tr>
+            <td><h5>lakestream.store</h5></td>
+            <td style="word-wrap: break-word;">(none)</td>
+            <td>String</td>
+            <td>Specifies the Lakestream Store implementation used to provide second-level data freshness for Paimon tables. Currently, only 'fluss' is supported.</td>
+        </tr>
         <tr>
             <td><h5>local-cache.block-size</h5></td>
             <td style="word-wrap: break-word;">1 mb</td>

diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html
@@ -968,6 +968,12 @@
             <td>Boolean</td>
             <td>Whether index file in data file directory.</td>
         </tr>
+        <tr>
+            <td><h5>lakestream.enabled</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Whether to enable lake-stream mode. When enabled, regular writes are routed to the Lakestream Store. The Lakestream Store's tiering service asynchronously writes data to Paimon. Reads from the original table name access only Paimon, while reads from the '$rt'-suffixed table name combine Paimon and Lakestream Store data.</td>
+        </tr>
         <tr>
             <td><h5>local-kv-db.block-size</h5></td>
             <td style="word-wrap: break-word;">4 kb</td>

diff --git a/docs/sidebars.js b/docs/sidebars.js
@@ -76,7 +76,8 @@ const sidebars = {
             "items": ["concepts/spec/tableindex", "concepts/spec/fileindex"]
           }
         ]
-      }
+      },
+      "concepts/lake-stream"
     ]
   },
   {
@@ -250,7 +251,8 @@ const sidebars = {
         "items": [
           "flink/sql-write",
           "flink/sql-query",
-          "flink/sql-lookup"
+          "flink/sql-lookup",
+          "flink/lake-stream"
         ]
       },
       {

diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -268,6 +268,18 @@ public InlineElement getDescription() {
     public static final ConfigOption<String> BRANCH =
             key("branch").stringType().defaultValue("main").withDescription("Specify branch name.");
 
+    public static final ConfigOption<Boolean> LAKE_STREAM_ENABLED =
+            key("lakestream.enabled")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Whether to enable lake-stream mode. When enabled, regular writes are "
+                                    + "routed to the Lakestream Store. The Lakestream Store's tiering "
+                                    + "service asynchronously writes data to Paimon. Reads from the "
+                                    + "original table name access only Paimon, while reads from the "
+                                    + "'$rt'-suffixed table name combine Paimon and Lakestream Store "
+                                    + "data.");
+
     public static final ConfigOption<Boolean> CHAIN_TABLE_ENABLED =
             key("chain-table.enabled")
                     .booleanType()

diff --git a/paimon-api/src/main/java/org/apache/paimon/options/CatalogOptions.java b/paimon-api/src/main/java/org/apache/paimon/options/CatalogOptions.java
@@ -255,4 +255,13 @@ public class CatalogOptions {
                             "Comma-separated list of file types to cache. "
                                     + "Supported values: meta, global-index, bucket-index, data, file-index, "
                                     + "or * for all of them.");
+
+    public static final ConfigOption<String> LAKE_STREAM_STORE =
+            ConfigOptions.key("lakestream.store")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "Specifies the Lakestream Store implementation used to provide "
+                                    + "second-level data freshness for Paimon tables. Currently, "
+                                    + "only 'fluss' is supported.");
 }
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
@@ -74,6 +74,7 @@
 import static org.apache.paimon.CoreOptions.PATH;
 import static org.apache.paimon.CoreOptions.TYPE;
 import static org.apache.paimon.catalog.CatalogUtils.checkNotBranch;
+import static org.apache.paimon.catalog.CatalogUtils.checkNotLakeStreamTable;
 import static org.apache.paimon.catalog.CatalogUtils.checkNotSystemDatabase;
 import static org.apache.paimon.catalog.CatalogUtils.checkNotSystemTable;
 import static org.apache.paimon.catalog.CatalogUtils.isSystemDatabase;
@@ -476,14 +477,16 @@ public void renameTable(Identifier fromTable, Identifier toTable, boolean ignore
         checkNotSystemTable(fromTable, "renameTable");
         checkNotSystemTable(toTable, "renameTable");
 
+        Table table;
         try {
-            getTable(fromTable);
+            table = getTable(fromTable);
         } catch (TableNotExistException e) {
             if (ignoreIfNotExists) {
                 return;
             }
             throw new TableNotExistException(fromTable);
         }
+        checkNotLakeStreamTable(table, fromTable, "renameTable");
 
         try {
             getTable(toTable);
@@ -536,6 +539,7 @@ public void replaceTable(Identifier identifier, Schema newSchema, boolean ignore
             }
             throw e;
         }
+        checkNotLakeStreamTable(existing, identifier, "replaceTable");
 
         TableType targetTableType = Options.fromMap(newSchema.options()).get(TYPE);
         if (!(existing instanceof FileStoreTable) || !targetTableType.equals(TableType.TABLE)) {

diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java b/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java
@@ -154,6 +154,22 @@ public static void checkNotBranch(Identifier identifier, String method) {
         }
     }
 
+    /**
+     * Rejects the operation on a lake-stream table: its identifier is what resolves the
+     * corresponding table in the Lakestream Store, and the store may still hold data that has not
+     * been tiered into Paimon.
+     */
+    public static void checkNotLakeStreamTable(Table table, Identifier identifier, String method) {
+        if (Options.fromMap(table.options()).get(CoreOptions.LAKE_STREAM_ENABLED)) {
+            throw new IllegalArgumentException(
+                    String.format(
+                            "Cannot '%s' for lake-stream table '%s', the table identifier is used "
+                                    + "to resolve the corresponding table in the Lakestream Store. "
+                                    + "Please set '%s' to false first.",
+                            method, identifier, CoreOptions.LAKE_STREAM_ENABLED.key()));
+        }
+    }
+
     public static void validateCreateTable(Schema schema, boolean dataTokenEnabled) {
         Options options = Options.fromMap(schema.options());
         checkArgument(