Repository navigation
[core][flink] Support lake-stream mode for Flink reads and writes - #10833
yunfengzhou-hub wants to merge 4 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.
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.
There was a problem hiding this comment.
🟡 Changes recommended
Critical routing gaps allow writes to bypass the store and UPDATE/DELETE scans to miss untiered rows or use stale Paimon values, risking data divergence and incorrect modifications.
3 open issues
What changed in this PR
Implements the core and Flink portions of PIP-39, keeping table history in Paimon while delegating fresh reads and regular writes to a pluggable lake-stream store.
Changes:
- Adds lake-stream options and catalog guards against renaming or replacing enabled tables.
- Introduces lazy store discovery,
$rtread routing, and store-backed Flink sinks. - Adds documentation, routing tests, and real Fluss integration coverage.
| File | Description |
|---|---|
pom.xml |
Defines Fluss integration-test dependency versions. |
paimon-flink/paimon-flink-common/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory |
Registers the fake store for discovery tests. |
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/LakeStreamTableFactoryTest.java |
Tests source/sink routing and validation. |
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/LakeStreamRoutingITCase.java |
Exercises routing through Flink SQL. |
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/LakeStreamFlussITCase.java |
Tests routing against a Fluss cluster. |
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/LakeStreamCatalogTest.java |
Tests discovery, lifecycle, and $rt resolution. |
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/FakeStoreTableFactory.java |
Records delegated factory contexts. |
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/FakeStoreSource.java |
Provides a bounded test source. |
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/FakeStoreSink.java |
Provides a test sink and records ability callbacks. |
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/FakeStoreData.java |
Stores test rows in memory. |
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/FakeStoreCatalogFactory.java |
Implements fake store discovery and configuration. |
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/FakeStoreCatalog.java |
Supplies controllable store metadata for tests. |
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lakestream/LakeStreamUtils.java |
Handles $rt identifiers and prefixed options. |
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lakestream/LakeStreamTableFactory.java |
Delegates source and sink creation to the store. |
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkTableFactory.java |
Adds lake-stream factory wiring. |
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalogFactory.java |
Passes Flink configuration into the catalog. |
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java |
Lazily discovers stores and resolves $rt tables. |
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java |
Selects Paimon or lake-stream routing. |
paimon-flink/paimon-flink-common/pom.xml |
Adds JDK-gated Fluss test dependencies. |
paimon-core/src/test/java/org/apache/paimon/catalog/CatalogTestBase.java |
Tests catalog guards and mode reversibility. |
paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java |
Guards REST rename and replace operations. |
paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java |
Guards JDBC rename operations. |
paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java |
Centralizes the lake-stream operation guard. |
paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java |
Guards shared rename and replace paths. |
paimon-api/src/main/java/org/apache/paimon/options/CatalogOptions.java |
Defines lakestream.store. |
paimon-api/src/main/java/org/apache/paimon/CoreOptions.java |
Defines lakestream.enabled. |
docs/sidebars.js |
Adds lake-stream documentation navigation. |
docs/generated/core_configuration.html |
Documents the table option. |
docs/generated/catalog_configuration.html |
Documents the catalog option. |
docs/docs/flink/lake-stream.md |
Explains Flink setup, routing, and Fluss usage. |
docs/docs/concepts/lake-stream.md |
Describes table semantics and lifecycle constraints. |
Review effort: Balanced
| if (isLakeStreamWrite(context)) { | ||
| return createLakeStreamSink(context); | ||
| } | ||
| return createPaimonTableSink(context, buildPaimonTable(context)); |
There was a problem hiding this comment.
Fixed. The write now reads lakestream.enabled from the schema persisted at path — the CatalogContext carries only path and branch, so connector options cannot restate it — and the read side is refused too: a $rt read or write through a connector mapping fails by name instead of being answered from Paimon, which a $rt read previously was. A plain-name read still returns Paimon snapshots, which is what the original name means on every path.
Covered by factory-level tests and an end-to-end case with a temporary table pointing at the table's directory, and the Flink page documents that a lake-stream table is read and written through the catalog.
| return sinkFactory.createDynamicTableSink( | ||
| delegatedContext(context, identifier, storeTable)); |
There was a problem hiding this comment.
Fixed by refusing the statement rather than routing the scan. LakeStreamDynamicTableSink wraps the store's sink and throws on applyRowLevelUpdate and applyRowLevelDelete: the scan resolves to Paimon snapshots while the modification goes to the store, so it could miss rows that have not been tiered or compute from a stale value.
SupportsOverwrite and SupportsDeletePushDown are withheld as well — a DELETE then falls back to the refused row-level path instead of being pushed into the store — while SupportsPartitioning is still forwarded, and LakeStreamSinkAbilityDriftTest keeps the withheld set in step with the Paimon and Fluss sinks. Which name should expose a view that is both readable and writable is a PIP-39 question, so that is left to a follow-up.
| FactoryUtil.discoverFactory( | ||
| Thread.currentThread().getContextClassLoader(), |
There was a problem hiding this comment.
Fixed. FlinkCatalog now retains the class loader FlinkCatalogFactory#createCatalog was given and uses it for both the SPI lookup and the delegated catalog context.
Covered by testStoreDiscoveryUsesTheClassLoaderSuppliedToTheCatalogFactory, which sets the thread context loader to one that cannot see the store's service registration while the supplied loader can.
A connector mapping that names only 'path' reads lakestream.enabled from the schema persisted there, so neither the mapping nor a query hint can turn the flag off and send the write to Paimon behind the store's back. With no catalog there is no store at all, so a '$rt' read and a '$rt' write are refused rather than answered from Paimon. Store discovery uses the class loader Flink supplied to the catalog factory instead of the calling thread's, for both the SPI lookup and the delegated catalog context. UPDATE and DELETE are refused: their scan resolves to Paimon snapshots while the modification is written to the store, so they could miss untiered rows or compute from stale values. Which name should expose a view that is both readable and writable is a PIP-39 question, left to a follow-up. LakeStreamSinkAbilityDriftTest keeps the sink's ability surface in step with the Paimon and Fluss sinks against an explicit list of withheld abilities, and trips when another store reaches the test class path.
JingsongLi
left a comment
JingsongLi
left a comment
There was a problem hiding this comment.
Requirement fit: SUPPORTED. Implementation: FINDINGS.
The Lake-stream routing has concrete end-to-end value and the earlier path-only bypass, row-level modification and discovery-classloader concerns are addressed. The normal JDK 8/Flink 1 package build passed. On JDK 11/Flink 1.20, the original targeted catalog, factory, SQL routing and real Fluss 1.0 mini-cluster tests passed; additional real executions reproduced the two issues below. The aligned two-column write and the complete-FileIO-options route serve as passing controls. Temporary reviewer changes were removed.
Release gate: BLOCK until the stored-value corruption and ordinary connector regression below are fixed. This review does not validate live Fluss tiering, stateful pipeline migration or a production rollout.
| } | ||
| ResolvedCatalogTable resolvedStoreTable = | ||
| new ResolvedCatalogTable( | ||
| (CatalogTable) storeTable, original.getCatalogTable().getResolvedSchema()); |
There was a problem hiding this comment.
[P1] Validate the Paimon/store row layout before delegating writes
The sink context combines the store table with Paimon's current resolved schema without checking their column order. The documentation and new catalog test explicitly allow Paimon-only column changes while the store catches up. With real Fluss 1.0, I created both tables as (a INT,b INT), inserted (1,2), ran ALTER TABLE orders MODIFY b INT FIRST only in Paimon, and inserted (4,3) (therefore b=4,a=3). SELECT a,b FROM orders$rt returned (1,2),(4,3) instead of (1,2),(3,4): the second row is silently stored under the wrong columns. Fluss's serializer maps names only when the column counts differ; equal-width rows are positional. Reject incompatible physical schemas/order before delegating, or explicitly align the emitted rows with the store schema, and cover the supported schema-drift interval with a real-store test.
There was a problem hiding this comment.
Fixed. Writes are now rejected before store sink creation unless the normalized Paimon and store schemas match in ordered physical fields and nullability, primary keys, partition keys, and shared immutable options. Store-private distribution settings and unrelated metadata are excluded.
| location.set(CoreOptions.BRANCH, options.get(CoreOptions.BRANCH)); | ||
| Table table = | ||
| FileStoreTableFactory.create( | ||
| CatalogContext.create(location, new FlinkFileIOLoader())); |
There was a problem hiding this comment.
[P2] Retain FileIO configuration while reading the persisted flag
This guard now runs for every connector-mode sink with a path, including ordinary tables, but its CatalogContext keeps only path/branch and drops the connection/FileIO options from the connector mapping. I reproduced this with the existing SPI-registered IsolatedDirectoryFileIO: a plain persisted table and mapping with the required root-dir successfully create a sink through the complete-options factory route, while the normal HEAD route throws Isolated file io requires option root-dir here before sink creation. The same loss affects storage plugins requiring endpoint/credential options supplied through the mapping. Preserve those options for FileIO construction while reading lakestream.enabled directly from the persisted schema, so a connector override still cannot restate the persisted flag.
There was a problem hiding this comment.
Fixed. The connector mapping's options are now used to construct the FileIO, while only path and branch are passed when loading the table. Therefore lakestream.enabled still comes exclusively from the persisted schema and cannot be overridden by DDL options or hints.
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.