[flink] add support for flink 2.3 - #3521
Conversation
bd9decf to
018112e
Compare
018112e to
5cb3dd2
Compare
9f3a6ba to
2856fc0
Compare
eb1da25 to
3f05e3f
Compare
|
Hi @sd4324530 and thank you for all the great work 🙏 Overall the PR LGTM 👍 However there is one thing I would like to clarify. Flink 2.3's new How do you think we should best handle this? Let me know your thoughts |
|
@polyzos I think this issue has always existed and isn't a new problem introduced with Flink 2.3. If the Fluss cluster's inherent mechanisms already prevent this problem, I think we should be able to disable this configuration by default. What do you think? |
|
@sd4324530 agreed 😄 @fresh-borzoni any thoughts since you are a heavy user? 😄 |
3f05e3f to
4728e8a
Compare
|
@polyzos @sd4324530 I'll take a look today, it's in my list to review 👍 |
leonardBang
left a comment
leonardBang
left a comment
There was a problem hiding this comment.
Thanks @sd4324530 for the contribution, I left two comments
| @Override | ||
| public SinkWriter<InputT> createWriter(WriterInitContext writerInitContext) throws IOException { | ||
| return createWriter( | ||
| writerInitContext.getMailboxExecutor(), writerInitContext.metricGroup()); |
There was a problem hiding this comment.
Would it make sense to keep the three-argument createWriter hook here and pass writerInitContext.getTaskInfo().getIndexOfThisSubtask(), as the Flink 2.2 adapter does? FlinkSink in the common module still overrides createWriter(MailboxExecutor, SinkWriterMetricGroup, int), while this adapter now invokes and declares only the two-argument method. When these classes are packaged together in the Flink 2.3 artifact, creating a writer can therefore fail with an AbstractMethodError.
Could we align the adapter signature with the common implementation? It may also be helpful to add a regression test that creates the sink writer through the final Flink 2.3 shaded artifact.
There was a problem hiding this comment.
This change originates from this commit:afc34bb
I'll fix it.
| private static void serializeMaterializedTableToCustomProperties( | ||
| CatalogMaterializedTable mt, Map<String, String> customProperties) { | ||
| // Serialize core materialized table properties | ||
| customProperties.put(MATERIALIZED_TABLE_DEFINITION_QUERY.key(), mt.getDefinitionQuery()); |
There was a problem hiding this comment.
Would it make sense to persist the original and expanded queries separately here? In Flink 2.3, getDefinitionQuery() represents the expanded query, while the restore path currently uses this single stored value for definitionQuery, originalQuery, and expandedQuery. If the original user SQL differs from the expanded query, a catalog round trip will therefore lose the original SQL.
Could we introduce dedicated properties for the original and expanded queries, while retaining the existing definition-query property as a compatibility fallback? It may also be helpful to add a round-trip test where the original and expanded queries are intentionally different.
a0a29b0 to
53ef1b9
Compare
🐛 Bug fix:
|
| Demo | Description | Result |
|---|---|---|
| 1 | Lookup Join: datagen events LEFT JOIN Fluss pk-table dim |
✅ 20/20 rows correctly enriched |
| 2 | Delta Join: two Fluss pk-tables INNER JOIN | ✅ Planner emits DeltaJoin operator; 5/5 rows correct, no join state |
| 3 | Cascaded Delta Join: three-level INNER JOIN | ✅ Planner emits two DeltaJoin operators; 5/5 rows correct |
Job graph excerpt (Demo 3):
Source: FlussSource-sales -> DropUpdateBefore
Source: FlussSource-sale_items -> DropUpdateBefore
Source: FlussSource-sale_item_addons -> DropUpdateBefore
DeltaJoin[7] -> Calc[8]
DeltaJoin[13] -> Calc[14] -> ConstraintEnforcer[15]
Sink(sale_item_addons_enriched): Writer
Patch
- Branch:
fightBoxing/fluss:fix-sink-adapter-flink-2.3 - Commit:
79ee7752e - Single-file change, only 6 insertions / 2 deletions.
Happy to open this as a follow-up PR against your support-flink-2.3 branch if you'd like — just let me know your preference. 🙏
fresh-borzoni
left a comment
fresh-borzoni
left a comment
There was a problem hiding this comment.
@sd4324530 Thank you for the PR, looks good overall, left a couple of comments, PTAL
| Map<String, String> options) { | ||
| // Validate required materialized table options first | ||
| String definitionQuery = options.get(MATERIALIZED_TABLE_DEFINITION_QUERY.key()); | ||
| String originalQuery = options.get(MATERIALIZED_TABLE_ORIGINAL_QUERY.key()); |
There was a problem hiding this comment.
Tables created before this PR only have definition-query, so both of these come back null and Flink 2.3 rejects them with checkNotNull therefore reading any existing materialized table fails with NullPointerException: Original query must not be null.
Should we fall back to definitionQuery when the new keys are absent?
There was a problem hiding this comment.
Hi @fresh-borzoni , I've read the design document for this feature(FLIP-546), and it should be compatible with existing materialized tables?
There was a problem hiding this comment.
@sd4324530
The case that's missing is upgrade, not new tables.
Someone creates a materialized table on Fluss today. Later they move to Flink 2.3 with this connector. That table was written before original-query and expanded-query existed, so it only has definition-query and every read of it now fails with NullPointerException: Original query must not be null.
Tables created after this PR are fine, since we write all three. It's the ones already out there that break. Try it :)
| * CatalogMaterializedTableAdapter} for Flink 1.20/2.2 treats original/expanded as no-ops, so | ||
| * the assertion deliberately targets only what the common adapter exposes: the consumed keys | ||
| * are absent from {@code getOptions()} and the {@code definitionQuery} matches the input. | ||
| * Verifying distinct original/expanded values survive a roundtrip is left to the Flink 2.3 |
There was a problem hiding this comment.
This says the roundtrip is covered in the 2.3 module, but I couldn't find that test there.
Could we add it, plus the case where neither key is present?
|
|
||
| assertThatThrownBy(() -> tEnv.explainSql(sql)) | ||
| .isInstanceOf(ValidationException.class) | ||
| .hasMessageContaining("doesn't support to do delta join optimization"); |
There was a problem hiding this comment.
Cascaded delta join works on 2.3. The same 3-way join with the sink PK aligned to the join key plans as nested DeltaJoin for me.
This test still throws only because its sink PK (c1, c2, c3) forces upsertMaterialize=[true].
Should we invert it?
| | Fluss Connector Versions | Supported Flink Versions | | ||
| |--------------------------|--------------------------| | ||
| | $FLUSS_VERSION_SHORT$ | 1.18, 1.19, 1.20 | | ||
| | $FLUSS_VERSION_SHORT$ | 1.18, 1.19, 1.20, 2.2, 2.3 | |
There was a problem hiding this comment.
TY, 2.2 was missing here. delta-joins.mdx still lists cascade joins as a 2.2 limitation and multi-way joins under "Future Plan", worth a Flink 2.3 section.
And fluss-flink-common/README.md's version table doesn't list 2.3.
WDYT?
| * <p>Restore the pre-2.3 behaviour by disabling the option in the SQL Gateway session so the | ||
| * refresh job can be scheduled and reach RUNNING. | ||
| */ | ||
| @BeforeAll |
There was a problem hiding this comment.
I removed this whole @BeforeAll and reverted the protected widening in the common class and 2/2 pass on 2.3. It also hides the parent's static setUp rather than extending it, so it will drift.
Can we drop it, or was there a failure that prompted it?
| // StreamPhysicalDeltaJoinForceValidator runs. Disable it here so the existing | ||
| // delta-join "doesn't support to do delta join optimization" error remains | ||
| // reachable from these tests. | ||
| tEnv.getConfig().set(ExecutionConfigOptions.TABLE_EXEC_SINK_REQUIRE_ON_CONFLICT, false); |
There was a problem hiding this comment.
Only the negative tests need this as they assert on an error message that Flink 2.3 changed.
The positive ones pass without it. Turning the validation off for the whole class to keep those assertions green also hides it from every other test here.
Could we update the assertions instead?
Partial upsert and delta join both plan fine on 2.3, I checked. What throws is re-keying That's FLIP-558 at the planner level though as it applies to every upsert sink on 2.3, not just Fluss. Our sink is a plain upsert sink and there's no connector hook to opt out, so there's nothing for us to fix. So I'd just document it in the docs. |
|
If I understand this correctly, the only difference between 2.2 module is the materialized table adapters, the rest is identical cc @loserwang1024 as I saw there was some discussion about 2.1 and 2.2 and it ended up with dropping 2.1, so there might be some context that I miss :) |
@fightBoxing Thank you for your review! |
Signed-off-by: Pei Yu <125331682@qq.com>
…Case
Flink 2.3 introduces ExecutionConfigOptions.TABLE_EXEC_SINK_REQUIRE_ON_CONFLICT
(table.exec.sink.require-on-conflict), defaulting to true. In
FlinkChangelogModeInferenceProgram, this triggers a ValidationException
("upsert key differs from primary key") before the
StreamPhysicalDeltaJoinForceValidator runs, so the Delta Join ITCases can no
longer reach the original "doesn't support to do delta join optimization"
error path.
Disable the option in Flink23DeltaJoinITCase#beforeEach so the existing
assertions remain valid. Production-side impact (real Fluss users hitting
this on multi-table joins / group-by + insert) is left to community
discussion.
Signed-off-by: Pei Yu <125331682@qq.com>
…Case
Flink 2.3 introduces ExecutionConfigOptions.TABLE_EXEC_SINK_REQUIRE_ON_CONFLICT
(table.exec.sink.require-on-conflict), defaulting to true. In
FlinkChangelogModeInferenceProgram, this triggers a ValidationException
("upsert key differs from primary key") before the partial-update handling
in FlinkTableSink#getSinkRuntimeProvider runs, so the partial upsert ITCases
(testPartialUpsert and testPartialUpsertDuringAddColumn) in
FlinkTableSinkITCase can no longer reach the Fluss sink layer.
Disable the option in Flink23TableSinkITCase#beforeEach so the existing
partial-upsert assertions remain valid.
Signed-off-by: Pei Yu <125331682@qq.com>
Signed-off-by: Pei Yu <125331682@qq.com>
Signed-off-by: Pei Yu <125331682@qq.com>
Signed-off-by: Pei Yu <125331682@qq.com>
Signed-off-by: Pei Yu <125331682@qq.com>
53ef1b9 to
e7bb9b1
Compare
Signed-off-by: Pei Yu <125331682@qq.com>
…RE_ON_CONFLICT`. Signed-off-by: Pei Yu <125331682@qq.com>
5 participants
Purpose
Linked issue: close #3520
This PR adds Flink 2.3 engine support to Apache Fluss by introducing a new
fluss-flink-2.3module. To accommodate the API changes inCatalogMaterializedTable/IntervalFreshnessintroduced in Flink 2.3, corresponding version adapters are introduced in the common module so that cross-version connector code can compile against multiple Flink majors.Brief change log
1. New
fluss-flink-2.3modulefluss-flink-2.3as a sub-module influss-flink/pom.xmlto provide connector support for Flink 2.3-based deployments.fluss-flink-2.2and ships with complete source and test directories. Version-specific adapters (e.g.,MultipleParameterToolAdapter,SchemaAdapter,SinkAdapter,TypeInformationAdapter) follow the existing pattern and are provided alongside the module.2. New version adapters in the common module
To handle the API changes of
CatalogMaterializedTable/IntervalFreshnessin Flink 2.3, the following adapters are introduced underfluss-flink-common/src/main/java/org/apache/fluss/flink/adapter/:CatalogMaterializedTableAdapter: wrapsCatalogMaterializedTable.Builderto abstract away differences introduced in Flink 2.3 (such as the neworiginalQuery/expandedQueryfields), so the shared common code does not depend on a specific Flink version.IntervalFreshnessAdapter: a new adapter forIntervalFreshnessand its innerTimeUnitenum. In Flink 2.3 theIntervalFreshness.TimeUnittype has been reworked (moved/repackaged), so this adapter provides a unified way to parse and serialize time units — converting betweenStringnames and the version-specificTimeUnitenum (via theTimeUnitAdapterwrapper). The common module no longer needs to depend on a Flink-version-specificIntervalFreshness.TimeUnitclass, and the existingMATERIALIZED_TABLE_INTERVAL_FRESHNESS_TIME_UNITconfig can keep itsstringType()form unchanged.3. Configuration and serialization compatibility
FlinkConnectorOptions.MATERIALIZED_TABLE_INTERVAL_FRESHNESS_TIME_UNITis changed fromenumType(IntervalFreshness.TimeUnit.class)tostringType(), avoiding a hard dependency on a Flink-version-specific enum class in the common module; concrete enum parsing is delegated toIntervalFreshnessAdapter.FlinkConversionsis updated to use the new adapters for materialization-table serialization/deserialization, ensuring consistent semantics across versions.4. Back-port to fluss-flink-2.2
CatalogMaterializedTableAdapteris also added tofluss-flink-2.2, andFlink22CatalogTestis updated accordingly, so that the 2.2 module continues to share the same common code after the new adapter is introduced.5. Test refactor: template-method pattern for version-specific constructors
The old
ResolvedCatalogMaterializedTableAdapter.create()helper masked the Flink-version-specificResolvedCatalogMaterializedTableconstructor signature with a 2-arg fake. To support Flink 2.3's new 5-arg constructor (which addsStartMode), the catalog test hierarchy is refactored to a template-method pattern:FlinkCatalogTest(influss-flink-common) now exposes aprotectedcreateResolvedCatalogMaterializedTable(...)method with a default 2-arg-constructor implementation.Flink22CatalogTestoverrides it to use the 4-arg constructor (origin,resolvedSchema,refreshMode,intervalFreshness).Flink23CatalogTestoverrides it to use the 5-arg constructor (additionally passingStartMode.of(StartMode.StartModeKind.FROM_BEGINNING)).This lets each Flink version exercise its native constructor signature, removes the need for the static helper, and makes the test code self-documenting about which Flink version it targets. Note that the parent
FlinkCatalogTestno longer importsResolvedCatalogMaterializedTableAdapter.6. Test coverage
Flink23*ITCase) is added under thefluss-flink-2.3module, covering catalog, metrics, procedure, authorization, sink, source (including binlog/changelog virtual tables, delta join, failover), and tiering.Flink23MultipleParameterToolTestis added to validateMultipleParameterToolAdapterbehavior.FlinkCatalogTestandFlink22CatalogTestare updated for the adapter-related cases.7. Test compatibility fix: Flink 2.3
ON CONFLICTvalidation vs. Delta Join testsWhile porting the Delta Join ITCases to Flink 2.3, an unexpected upstream planner behavior change was discovered. This PR works around it in the test code only; the broader impact on Fluss end-users upgrading to Flink 2.3 is left for community discussion.
What changed in Flink 2.3. Flink 2.3 introduces a new planner option
ExecutionConfigOptions.TABLE_EXEC_SINK_REQUIRE_ON_CONFLICT(table.exec.sink.require-on-conflict, defaulttrue). InsideFlinkChangelogModeInferenceProgram.SatisfyUpdateKindTraitVisitor.analyzeUpsertMaterializeStrategy, Flink now throwsValidationException("The query has an upsert key that differs from the primary key of the sink table ...")whenever:TABLE_EXEC_SINK_UPSERT_MATERIALIZE = AUTO(default), andON CONFLICTclause is supplied.In Flink 2.2 this validation did not exist.
Why this affects Fluss's Delta Join tests. Delta Join semantics define the upsert key from the join condition, not from the sink's primary key. In several Flink23DeltaJoinITCase scenarios (e.g.,
testDeltaJoinWithJoinKeyExceedsPrimaryKeywith join conditionc1=c2 AND d1=d2 AND e1=e2into a sink whose PK is(c1, d1)), the upsert key(c1, d1, e1)legitimately exceeds the sink PK. Under Flink 2.2 these tests asserted thatStreamPhysicalDeltaJoinForceValidatorthrowsThe current sql doesn't support to do delta join optimization. Under Flink 2.3 the newON CONFLICTvalidator fires before the Delta Join validator is reached, so the original error message is never produced and the assertions fail.Why this PR only touches the tests. The new Flink validation is the correct behavior in the general case — silently allowing mismatched upsert keys leads to non-deterministic results at the sink, which is exactly what Flink 2.3 is trying to prevent. Disabling it in the connector would silently regress that protection for all Fluss users. Whether/how Fluss should expose this knob (e.g., as a Fluss-level connector option, or by injecting a
ConflictStrategyfromFlinkTableSinkwhen the underlying Fluss table has alast_rowmerge engine) is a product-level decision that should be discussed with the community and is out of scope for this PR.This is the minimal, scoped workaround.
Flink22DeltaJoinITCaseis untouched (the option did not exist in 2.2).8. Test compatibility fix: Flink 2.3
ON CONFLICTvalidation vs. Table Sink partial-upsert testsA second, distinct impact of the same Flink 2.3 validation was uncovered while running
Flink23TableSinkITCaseon CI: the Fluss partial-upsert test path is also blocked. Same planner option, same root cause, but a different surface — addressed with the same minimal-scoped pattern.The fix.
Flink23TableSinkITCasewas previously an empty subclass ofFlinkTableSinkITCase. This PR turns it into a proper subclass with a single@BeforeEachthat disablesTABLE_EXEC_SINK_REQUIRE_ON_CONFLICTon the streamingTableConfig, mirroring the precedent set in Section 7.Why this PR only touches the tests. Same reasoning as Section 7: Flink 2.3's validation is the correct general-case behavior, and silently turning it off at the connector level would regress the protection for real users. The proper follow-up is for Fluss to participate in the new model — most naturally by having
FlinkTableSink.applyOperations(...)(orgetSinkRuntimeProvider) inject aConflictStrategy(e.g.,DEDUPLICATE, semantically equivalent to Fluss's current first-write-wins plus target-column overwrite) when the sink is a Fluss PK table, so partial upserts remain expressible from SQL. That is a product/API change and is explicitly out of scope for this engine-port PR.Flink22TableSinkITCaseis untouched (the option did not exist in 2.2).Tests
Flink23MultipleParameterToolTestFlink23CatalogTestFlink23TieringCommitOperatorTestFlinkCatalogTest,Flink22CatalogTest(adapter-related cases updated;Flink22/23CatalogTestnow exercise their nativeResolvedCatalogMaterializedTableconstructors via the new template-method override)fluss-flink-2.3module):Flink23CatalogITCase,Flink23MaterializedTableITCaseFlink23MetricsITCaseFlink23ProcedureITCaseFlink23AuthorizationITCaseFlink23ComplexTypeITCase,Flink23TableSinkITCase(now works around Flink 2.3's newON CONFLICTplanner validation for partial upserts, see Section 8),Flink23UndoRecoveryITCaseFlink23BinlogVirtualTableITCase,Flink23ChangelogVirtualTableITCase,Flink23DeltaJoinITCase(now works around Flink 2.3's newON CONFLICTplanner validation, see Section 7),Flink23TableSourceBatchITCase,Flink23TableSourceFailOverITCase,Flink23TableSourceITCaseFlink23TieringITCaseAPI and Format
No breaking changes to the public API.
Documentation
Generative AI disclosure