Repository navigation
[core][flink] Support lake-stream mode for Flink reads and writes - #10833
Open
yunfengzhou-hub wants to merge 2 commits into
Open
yunfengzhou-hub wants to merge 2 commits into
yunfengzhou-hub wants to merge 2 commits into
Conversation
Implements PIP-39 for Flink together with the engine-neutral table-format contract it depends on: a Paimon table keeps its history in Paimon while a pluggable Lakestream Store serves its most recent writes, so a query can return data that is seconds old. Table-format contract: - 'lakestream.store' (catalog option) names the store implementation, and 'lakestream.enabled' (table option) marks a table as a lake-stream table. The latter stays mutable, so the mode can be turned off again with ALTER TABLE instead of by dropping the table. - renameTable and replaceTable are rejected while the option is true. The identifier is what resolves the store's table, so a rename breaks that lookup, and a replace can discard data the store has not tiered yet or drop the option without the store being told. Column changes stay allowed and still update only the Paimon table. The guards sit in AbstractCatalog, and additionally in JdbcCatalog and RESTCatalog, which override those methods without delegating to super. Flink: - FlinkCatalog itself carries lake-stream mode: it owns the store catalog in one nullable field, resolves the '$rt' name in the getTable funnel that already dispatches Paimon's other '$'-suffixed system tables, and hands a LakeStreamTableFactory to the table factory it returns. A catalog that does not set 'lakestream.store' keeps the field null and behaves as it does today. - getOrCreateLakeStreamStoreCatalog creates the store catalog on first use rather than when the catalog is created, so a store that is unreachable or absent from the classpath does not stop this catalog from resolving table metadata: reading a lake-stream table by its own name reads Paimon and must not depend on the store, and turning the mode off again with ALTER TABLE has to stay possible. The result is cached and closed with the catalog. - The store is discovered through Flink's CatalogFactory SPI using 'lakestream.store' as the identifier, with the catalog options prefixed with '<identifier>.' passed after prefix removal, so Paimon compiles against no store class. - Reading '<table>$rt' resolves the base identifier against the store, which serves the read. The '$rt' name is read-only and rejects a time-travel timestamp. - A regular write to a lake-stream table is delegated to the store's sink. Query hints cannot move a write between Paimon and the store: Flink merges a hint into the table it hands the factory, overwriting the persisted value there, so both routing paths read the option from the catalog instead. - A write to a lake-stream table planned by a catalog that sets no 'lakestream.store', which is also what using Paimon as a plain connector gives, is rejected instead of quietly written to Paimon: the store would keep serving the same table's recent data and the two would then hold different values for the same keys. - LakeStreamTableFactory builds the store's source and sink and is a real DynamicTableSourceFactory and DynamicTableSinkFactory. AbstractFlinkTableFactory holds it rather than Catalog#getFactory returning it, so a catalog still hands out a FlinkTableFactory. Each of the base's two entry points reads 'if this statement belongs to lake-stream mode, build its source or sink there, else build Paimon's', and every one of those four halves is a protected method. - LakeStreamUtils holds the '$rt' naming rule and the store option prefix rule, which the catalog and the table factory both apply. INSERT OVERWRITE is not routed to Paimon: a write is served by the store, so an overwrite reaches the store's sink and fails there.
yunfengzhou-hub
force-pushed
the
pip-39-flink-lake-stream
branch
from
October 27, 2026 10:40
a3e3036 to
6cc5e5d
Compare
Fluss ships Java 11 bytecode, so javac 8 cannot compile LakeStreamFlussITCase, and the JUnit extension that fluss-test-utils registers through the service loader is auto-detected by every test in the module once that jar reaches the test class path. Gate the dependencies and the test source on JDK 11.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Purpose
Closes #10692 and closes #10693, part of #10690.
Implements PIP-39 for Flink, plus the engine-neutral table-format contract it depends on. A lake-stream table keeps its history in Paimon while a pluggable Lakestream Store serves its most recent writes, so a query can return data that is seconds old instead of waiting for the next snapshot. Fluss is the first store.
The catalog option
lakestream.storenames the store. The table optionlakestream.enabledmarks a table as a lake-stream table and stays mutable, soALTER TABLEcan turn the mode off. While it istrue,renameTableandreplaceTableare rejected, because both break the store's side of the table;JdbcCatalogandRESTCatalogoverride those methods without calling super, so the rejection has to be repeated in each.The store is discovered through Flink's own
CatalogFactorySPI, with the catalog options prefixed with<identifier>.forwarded after the prefix is stripped, so Paimon compiles against no store class. It is created on first use rather than together with the catalog, so an unreachable store still leaves the catalog able to resolve table metadata and to turn the mode off again.Reading
my_table$rtis served by the store, readingmy_tablekeeps returning Paimon snapshots, and a regular write goes to the store's sink. A query hint cannot move a write between Paimon and the store, and a write planned by a catalog that names no store is rejected rather than quietly written to Paimon, which would leave the two holding different values for the same keys.Adds
docs/docs/concepts/lake-stream.mdfor the table semantics anddocs/docs/flink/lake-stream.mdfor the Flink configuration, the SQL surface and a Fluss walkthrough. A Flink CDC pipeline does not plan through the SQL sink, so it still writes straight to Paimon; the Flink page documents the precaution.Tests
CatalogTestBasegains cases for the identity guards, so they run for every catalog implementation and not only the filesystem one.LakeStreamCatalogTestcovers store discovery and the$rtinterception, including branch suffixes, Paimon system tables, time travel and a missing store-side table.LakeStreamTableFactoryTestcovers source and sink routing, query hints, the format-table rejection, and the missing-store rejection in both catalog and connector mode.LakeStreamRoutingITCaseruns the behaviour end to end over SQL in batch and in streaming mode, against a fake store registered through the sameCatalogFactorySPI a real store uses.LakeStreamFlussITCaseruns it against a real Fluss mini cluster.